Make ReadableComponent expose ByteCursors

Motivation:
Another way to process the readable data in a buffer.
This might be faster for composite buffers, since their byte cursors are a bit slower than the MemSegBuffer due to the indirections and more complicated logic.

Modification:
ReadableComponent now have openCursor method.

Note that we *don't* add an openReverseCursor method on ReadableComponent.
The reason is that forEachReadable iterates the components in the forward direction, and it's really confusing to then iterate the bytes in a backwards direction.
Working with both directions at the same time is very error prone.

Result:
It is now possible to process readable components with byte cursors.

Author: chrisvest

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

@@ -72,5 +72,18 @@ public interface ReadableComponent {
      * @return A new {@link ByteBuffer}, with its own position and limit, for this memory component.
      */
     ByteBuffer readableBuffer();
+
+    /**
+     * Open a cursor to iterate the readable bytes of this component.
+     * Any offsets internal to the component are not modified by the cursor.
+     * <p>
+     * Care should be taken to ensure that the buffers lifetime extends beyond the cursor and the iteration, and that
+     * the internal offsets of the component (such as {@link Buffer#readerOffset()} and {@link Buffer#writerOffset()})
+     * are not modified while the iteration takes place. Otherwise unpredictable behaviour might result.
+     *
+     * @return A {@link ByteCursor} for iterating the readable bytes of this buffer.
+     * @see Buffer#openCursor()
+     */
+    ByteCursor openCursor();
     // todo for Unsafe-based impl, DBB.attachment needs to keep underlying memory alive
 }

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

@@ -300,6 +300,11 @@ public void copyInto(int srcPos, Buffer dest, int destPos, int length) {
         }
     }
 
+    @Override
+    public ByteCursor openCursor() {
+        return openCursor(readerOffset(), readableBytes());
+    }
+
     @Override
     public ByteCursor openCursor(int fromOffset, int length) {
         if (seg == CLOSED_SEGMENT) {
@@ -364,6 +369,12 @@ public int bytesLeft() {
         };
     }
 
+    @Override
+    public ByteCursor openReverseCursor() {
+        int woff = writerOffset();
+        return openReverseCursor(woff == 0? 0 : woff - 1, readableBytes());
+    }
+
     @Override
     public ByteCursor openReverseCursor(int fromOffset, int length) {
         if (seg == CLOSED_SEGMENT) {

src/test/java/io/netty/buffer/api/BufferTest.java

@@ -2890,6 +2890,39 @@ public void forEachReadableMustAllowCollectingBuffersInArray(Fixture fixture) {
         }
     }
 
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void forEachReadableMustExposeByteCursors(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(32).order(BIG_ENDIAN)) {
+            buf.writeLong(0x0102030405060708L);
+            buf.writeLong(0x1112131415161718L);
+            assertEquals(0x01020304, buf.readInt());
+            try (Buffer actualData = allocator.allocate(buf.readableBytes()).order(BIG_ENDIAN);
+                 Buffer expectedData = allocator.allocate(12).order(BIG_ENDIAN)) {
+                expectedData.writeInt(0x05060708);
+                expectedData.writeInt(0x11121314);
+                expectedData.writeInt(0x15161718);
+
+                buf.forEachReadable(0, (i, component) -> {
+                    ByteCursor forward = component.openCursor();
+                    while (forward.readLong()) {
+                        actualData.writeLong(forward.getLong());
+                    }
+                    while (forward.readByte()) {
+                        actualData.writeByte(forward.getByte());
+                    }
+                    return true;
+                });
+
+                assertEquals(expectedData.readableBytes(), actualData.readableBytes());
+                while (expectedData.readableBytes() > 0) {
+                    assertEquals(expectedData.readByte(), actualData.readByte());
+                }
+            }
+        }
+    }
+
     @ParameterizedTest
     @MethodSource("nonCompositeAllocators")
     public void forEachWritableMustVisitBuffer(Fixture fixture) {