Add Buf.forEachWritable

Pass iteration indexes through.

Author: chrisvest

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

@@ -17,7 +17,6 @@
 
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
-import java.util.function.Consumer;
 
 /**
  * A reference counted buffer of memory, with separate reader and writer offsets.
@@ -490,19 +489,74 @@ default void ensureWritable(int size) {
      *
      * @return The number of components in this buffer.
      */
-    int componentCount();
+    int countComponents();
 
     /**
-     * Process all readable components of this buffer, and return the number of components consumed.
+     * Get the number of "components" in this buffer, that are readable. These are the components that would be
+     * processed by {@link #forEachReadable(int, ComponentProcessor)}. For composite buffers, this is the number of
+     * transitive constituent buffers that are readable, while non-composite buffers only have at most one readable
+     * component.
      * <p>
-     * The number of components consumed may be less than the {@linkplain #componentCount() component count} if not all
-     * of them have readable data.
+     * The number of readable components may be less than the {@link #countComponents() component count}, if not all of
+     * them have readable data.
      *
+     * @return The number of readable components in this buffer.
+     */
+    int countReadableComponents();
+
+    /**
+     * Get the number of "components" in this buffer, that are writable. These are the components that would be
+     * processed by {@link #forEachWritable(int, ComponentProcessor)}. For composite buffers, this is the number of
+     * transitive constituent buffers that are writable, while non-composite buffers only have at most one writable
+     * component.
+     * <p>
+     * The number of writable components may be less than the {@link #countComponents() component count}, if not all of
+     * them have space for writing.
+     *
+     * @return The number of writable components in this buffer.
+     */
+    int countWritableComponents();
+
+    /**
+     * Process all readable components of this buffer, and return the number of components processed.
+     * <p>
+     * The given {@linkplain ComponentProcessor processor} is called for each component in this buffer, and passed a
+     * component index, for the given component in the iteration, and a {@link Component} object for accessing the data
+     * within the given component.
+     * <p>
+     * The component index is specific to the particular invokation of this method, and may change. The first call to
+     * the consumer will be passed the given initial index, and the next call will be passed the initial index plus one,
+     * and so on.
+     * <p>
+     * The {@link ComponentProcessor} may stop the iteration at any time by returning {@code false}. This may cause the
+     * number of components processed to be returned as a negative number (to signal early return), and the number of
+     * components processed may then be less than the {@linkplain #countReadableComponents() readable component count}.
+     * <p>
      * <strong>Note</strong> that the {@link Component} instance passed to the consumer could be reused for multiple
      * calls, so the data must be extracted from the component in the context of the iteration.
+     * <p>
+     * The {@link ByteBuffer} instances obtained from the component, share life time with that internal component.
+     * This means they can be accessed as long as the internal memory store remain unchanged. Methods that may cause
+     * such changes, are any method that requires the buffer to be {@linkplain #isOwned() owned}.
+     * <p>
+     * The best way to ensure this doesn't cause any trouble, is to use the buffers directly as part of the iteration,
+     * or immediately after the iteration.
+     * <p>
+     * <strong>Note</strong> that the arrays, memory addresses, and byte buffers exposed as components by this method,
+     * should not be used for changing the buffer contents. Doing so may cause undefined behaviour.
+     * <p>
+     * Changes to position and limit of the byte buffers exposed via the processed components, are not reflected back to
+     * this buffer instance.
      *
-     * @param consumer The consumer that will be used to process the buffer components.
-     * @return The number of readable components processed, which may be less than {@link #componentCount()}.
+     * @param initialIndex The initial index of the iteration, and the index that will be passed to the first call to
+     *                    the {@linkplain ComponentProcessor#process(int, Component) processor}.
+     * @param processor The processor that will be used to process the buffer components.
+     * @return The number of readable components processed, as a positive number of all readable components were
+     * processed, or as a negative number if the iteration was stopped because
+     * {@link ComponentProcessor#process(int, Component)} returned {@code false}.
+     * In any case, the number of components processed may be less than {@link #countComponents()}.
      */
-    int forEachReadable(Consumer<Component> consumer);
+    int forEachReadable(int initialIndex, ComponentProcessor processor);
+
+    int forEachWritable(int initialIndex, ComponentProcessor processor);
 }

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

@@ -16,10 +16,10 @@
 package io.netty.buffer.api;
 
 import java.nio.ByteBuffer;
-import java.util.function.Consumer;
 
 /**
- * A view onto the buffer component being processed in a given iteration of {@link Buf#forEachReadable(Consumer)}.
+ * A view onto the buffer component being processed in a given iteration of
+ * {@link Buf#forEachReadable(int, ComponentProcessor)}, or {@link Buf#forEachWritable(int, ComponentProcessor)}.
  * <p>
  * Instances of this interface are allowed to be mutable behind the scenes, and the data is only guaranteed to be
  * consistent within the given iteration.
@@ -28,30 +28,44 @@ public interface Component {
 
     /**
      * Check if this component is backed by a cached byte array than can be accessed cheaply.
+     * <p>
+     * <strong>Note</strong> that regardless of what this method returns, the array should not be used to modify the
+     * contents of this buffer component.
      *
      * @return {@code true} if {@link #array()} is a cheap operation, otherwise {@code false}.
      */
-    boolean hasCachedArray();
+    boolean hasArray();
 
     /**
      * Get a byte array of the contents of this component.
      * <p>
      * <strong>Note</strong> that the array is meant to be read-only. It may either be a direct reference to the
      * concrete array instance that is backing this component, or it is a fresh copy.
+     * Writing to the array may produce undefined behaviour.
      *
      * @return A byte array of the contents of this component.
      */
     byte[] array();
 
+    int arrayOffset();
+
     /**
      * Give the native memory address backing this buffer, or return 0 if this buffer has no native memory address.
+     * <p>
+     * <strong>Note</strong> that the address should not be used for writing to the buffer memory, and doing so may
+     * produce undefined behaviour.
+     *
      * @return The native memory address, if any, otherwise 0.
      */
     long nativeAddress();
 
     /**
-     * Build a {@link ByteBuffer} instance for this memory component.
+     * Get a {@link ByteBuffer} instance for this memory component.
+     * <p>
+     * <strong>Note</strong> that the {@link ByteBuffer} is read-only, to prevent write accesses to the memory,
+     * when the buffer component is obtained through {@link Buf#forEachReadable(int, ComponentProcessor)}.
+     *
      * @return A new {@link ByteBuffer} for this memory component.
      */
-    ByteBuffer byteBuffer();
+    ByteBuffer buffer();
 }

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

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2020 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer.api;
+
+@FunctionalInterface
+public interface ComponentProcessor {
+    boolean process(int index, Component component);
+}

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

@@ -19,7 +19,6 @@
 import java.nio.ByteOrder;
 import java.util.Arrays;
 import java.util.Objects;
-import java.util.function.Consumer;
 
 final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
     /**
@@ -49,7 +48,12 @@ final class CompositeBuf extends RcSupport<Buf, CompositeBuf> implements Buf {
     private boolean readOnly;
 
     CompositeBuf(Allocator allocator, Buf[] bufs) {
-        this(allocator, true, bufs.clone(), COMPOSITE_DROP); // Clone prevents external modification of array.
+        this(allocator, true, filterExternalBufs(bufs), COMPOSITE_DROP);
+    }
+
+    private static Buf[] filterExternalBufs(Buf[] bufs) {
+        // Allocating a new array unconditionally also prevents external modification of the array.
+        return Arrays.stream(bufs).filter(b -> b.capacity() > 0).toArray(Buf[]::new);
     }
 
     private CompositeBuf(Allocator allocator, boolean isSendable, Buf[] bufs, Drop<CompositeBuf> drop) {
@@ -617,7 +621,14 @@ void extendWith(Buf extension) {
                     (extension.readOnly()? "read-only." : "writable."));
         }
 
-        long newSize = capacity() + (long) extension.capacity();
+        long extensionCapacity = extension.capacity();
+        if (extensionCapacity == 0) {
+            // Extending by a zero-sized buffer makes no difference. Especially since it's not allowed to change the
+            // capacity of buffers that are constiuents of composite buffers.
+            return;
+        }
+
+        long newSize = capacity() + extensionCapacity;
         Allocator.checkSize(newSize);
 
         Buf[] restoreTemp = bufs; // We need this to restore our buffer array, in case offset computations fail.
@@ -702,21 +713,63 @@ public void compact() {
     }
 
     @Override
-    public int componentCount() {
+    public int countComponents() {
         int sum = 0;
         for (Buf buf : bufs) {
-            sum += buf.componentCount();
+            sum += buf.countComponents();
         }
         return sum;
     }
 
     @Override
-    public int forEachReadable(Consumer<Component> consumer) {
+    public int countReadableComponents() {
+        int sum = 0;
+        for (Buf buf : bufs) {
+            sum += buf.countReadableComponents();
+        }
+        return sum;
+    }
+
+    @Override
+    public int countWritableComponents() {
+        int sum = 0;
+        for (Buf buf : bufs) {
+            sum += buf.countWritableComponents();
+        }
+        return sum;
+    }
+
+    @Override
+    public int forEachReadable(int initialIndex, ComponentProcessor processor) {
         checkReadBounds(readerOffset(), Math.max(1, readableBytes()));
         int visited = 0;
         for (Buf buf : bufs) {
             if (buf.readableBytes() > 0) {
-                visited += buf.forEachReadable(consumer);
+                int count = buf.forEachReadable(visited + initialIndex, processor);
+                if (count > 0) {
+                    visited += count;
+                } else {
+                    visited = -visited + count;
+                    break;
+                }
+            }
+        }
+        return visited;
+    }
+
+    @Override
+    public int forEachWritable(int initialIndex, ComponentProcessor processor) {
+        checkWriteBounds(writerOffset(), Math.max(1, writableBytes()));
+        int visited = 0;
+        for (Buf buf : bufs) {
+            if (buf.writableBytes() > 0) {
+                int count = buf.forEachWritable(visited + initialIndex, processor);
+                if (count > 0) {
+                    visited += count;
+                } else {
+                    visited = -visited + count;
+                    break;
+                }
             }
         }
         return visited;