Merge pull request #26 from netty/update-examples

Update examples

Author: chrisvest

src/main/java/io/netty/buffer/api/Buf.java

@@ -562,7 +562,7 @@ default void ensureWritable(int size) {
      * {@link ReadableComponentProcessor#process(int, ReadableComponent)} returned {@code false}.
      * In any case, the number of components processed may be less than {@link #countComponents()}.
      */
-    int forEachReadable(int initialIndex, ReadableComponentProcessor processor);
+    <E extends Exception> int forEachReadable(int initialIndex, ReadableComponentProcessor<E> processor) throws E;
 
     /**
      * Process all writable components of this buffer, and return the number of components processed.
@@ -601,5 +601,5 @@ default void ensureWritable(int size) {
      * {@link WritableComponentProcessor#process(int, WritableComponent)} returned {@code false}.
      * In any case, the number of components processed may be less than {@link #countComponents()}.
      */
-    int forEachWritable(int initialIndex, WritableComponentProcessor processor);
+    <E extends Exception> int forEachWritable(int initialIndex, WritableComponentProcessor<E> processor) throws E;
 }

src/main/java/io/netty/buffer/api/ComponentProcessor.java

@@ -26,7 +26,7 @@ public interface ComponentProcessor {
      * A processor of {@linkplain ReadableComponent readable components}.
      */
     @FunctionalInterface
-    interface ReadableComponentProcessor extends ComponentProcessor {
+    interface ReadableComponentProcessor<E extends Exception> extends ComponentProcessor {
         /**
          * Process the given component at the given index in the
          * {@link Buf#forEachReadable(int, ReadableComponentProcessor) iteration}.
@@ -41,14 +41,14 @@ interface ReadableComponentProcessor extends ComponentProcessor {
          * @return {@code true} if the iteration should continue and more components should be processed, otherwise
          * {@code false} to stop the iteration early.
          */
-        boolean process(int index, ReadableComponent component);
+        boolean process(int index, ReadableComponent component) throws E;
     }
 
     /**
      * A processor of {@linkplain WritableComponent writable components}.
      */
     @FunctionalInterface
-    interface WritableComponentProcessor extends ComponentProcessor {
+    interface WritableComponentProcessor<E extends Exception> extends ComponentProcessor {
         /**
          * Process the given component at the given index in the
          * {@link Buf#forEachWritable(int, WritableComponentProcessor)}  iteration}.
@@ -63,7 +63,7 @@ interface WritableComponentProcessor extends ComponentProcessor {
          * @return {@code true} if the iteration should continue and more components should be processed, otherwise
          * {@code false} to stop the iteration early.
          */
-        boolean process(int index, WritableComponent component);
+        boolean process(int index, WritableComponent component) throws E;
     }
 
     /**

src/main/java/io/netty/buffer/api/CompositeBuf.java

@@ -801,7 +801,8 @@ public int countWritableComponents() {
     }
 
     @Override
-    public int forEachReadable(int initialIndex, ReadableComponentProcessor processor) {
+    public <E extends Exception> int forEachReadable(int initialIndex, ReadableComponentProcessor<E> processor)
+            throws E {
         checkReadBounds(readerOffset(), Math.max(1, readableBytes()));
         int visited = 0;
         for (Buf buf : bufs) {
@@ -819,7 +820,8 @@ public int forEachReadable(int initialIndex, ReadableComponentProcessor processo
     }
 
     @Override
-    public int forEachWritable(int initialIndex, WritableComponentProcessor processor) {
+    public <E extends Exception> int forEachWritable(int initialIndex, WritableComponentProcessor<E> processor)
+            throws E {
         checkWriteBounds(writerOffset(), Math.max(1, writableBytes()));
         int visited = 0;
         for (Buf buf : bufs) {

src/main/java/io/netty/buffer/api/memseg/MemSegBuf.java

@@ -548,13 +548,15 @@ public int countWritableComponents() {
     }
 
     @Override
-    public int forEachReadable(int initialIndex, ReadableComponentProcessor processor) {
+    public <E extends Exception> int forEachReadable(int initialIndex, ReadableComponentProcessor<E> processor)
+            throws E {
         checkRead(readerOffset(), Math.max(1, readableBytes()));
         return processor.process(initialIndex, this)? 1 : -1;
     }
 
     @Override
-    public int forEachWritable(int initialIndex, WritableComponentProcessor processor) {
+    public <E extends Exception> int forEachWritable(int initialIndex, WritableComponentProcessor<E> processor)
+            throws E {
         checkWrite(writerOffset(), Math.max(1, writableBytes()));
         return processor.process(initialIndex, this)? 1 : -1;
     }

src/test/java/io/netty/buffer/api/examples/FileCopyExample.java

@@ -19,7 +19,6 @@
 import io.netty.buffer.api.Buf;
 import io.netty.buffer.api.Send;
 
-import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.Path;
 import java.util.concurrent.ArrayBlockingQueue;
@@ -34,27 +33,23 @@
 public final class FileCopyExample {
     public static void main(String[] args) throws Exception {
         ExecutorService executor = Executors.newFixedThreadPool(2);
-        ArrayBlockingQueue<Object> queue = new ArrayBlockingQueue<>(8);
-        Object done = new Object();
+        ArrayBlockingQueue<Send<Buf>> queue = new ArrayBlockingQueue<>(8);
         try (Allocator allocator = Allocator.pooledDirect();
              var input = FileChannel.open(Path.of("/dev/urandom"), READ);
              var output = FileChannel.open(Path.of("random.bin"), CREATE, TRUNCATE_EXISTING, WRITE)) {
+            Send<Buf> done = allocator.compose().send();
 
             var reader = executor.submit(() -> {
-                var buf = ByteBuffer.allocateDirect(1024);
                 for (int i = 0; i < 1024; i++) {
-                    buf.clear();
-                    while (buf.hasRemaining()) {
-                        int r = input.read(buf);
-                        System.out.println("r = " + r);
-                        System.out.println("buf = " + buf);
-                    }
-                    buf.flip();
                     try (Buf in = allocator.allocate(1024)) {
                         System.out.println("in = " + in);
-                        while (buf.hasRemaining()) {
-                            in.writeByte(buf.get());
-                        }
+                        in.forEachWritable(0, (index, component) -> {
+                            var bb = component.writableBuffer();
+                            while (bb.hasRemaining()) {
+                                input.read(bb);
+                            }
+                            return true;
+                        });
                         System.out.println("Sending " + in.readableBytes() + " bytes.");
                         queue.put(in.send());
                     }
@@ -64,19 +59,17 @@ public static void main(String[] args) throws Exception {
             });
 
             var writer = executor.submit(() -> {
-                var buf = ByteBuffer.allocateDirect(1024);
-                Object msg;
-                while ((msg = queue.take()) != done) {
-                    buf.clear();
-                    @SuppressWarnings("unchecked")
-                    Send<Buf> send = (Send<Buf>) msg;
+                Send<Buf> send;
+                while ((send = queue.take()) != done) {
                     try (Buf out = send.receive()) {
                         System.out.println("Received " + out.readableBytes() + " bytes.");
-                        out.copyInto(0, buf, 0, out.readableBytes());
-                        buf.position(0).limit(out.readableBytes());
-                    }
-                    while (buf.hasRemaining()) {
-                        output.write(buf);
+                        out.forEachReadable(0, (index, component) -> {
+                            var bb = component.readableBuffer();
+                            while (bb.hasRemaining()) {
+                                output.write(bb);
+                            }
+                            return true;
+                        });
                     }
                 }
                 output.force(true);