First draft of const buffers and how to have buffers as constants

Currently only has a strawman implementation.

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

@@ -128,6 +128,14 @@ import java.nio.ByteOrder;
  * perhaps unknown, piece of code, and relinquish your ownership of that buffer region in the process.
  * Examples include aggregating messages into an accumulator buffer, and sending messages down the pipeline for
  * further processing, as split buffer regions, once their data has been received in its entirety.
+ *
+ * <h3>Buffers as constants</h3>
+ *
+ * Sometimes, the same bit of data will be processed or transmitted over and over again. In such cases, it can be
+ * tempting to allocate and fill a buffer once, and then reuse it.
+ * Such reuse must be done carefully, however, to avoid a number of bugs.
+ * The {@link BufferAllocator} has a {@link BufferAllocator#constBufferSupplier(byte[])} method that solves this, and
+ * prevents these bugs from occurring.
  */
 public interface Buffer extends Rc<Buffer>, BufferAccessors {
     /**
@@ -308,6 +316,23 @@ public interface Buffer extends Rc<Buffer>, BufferAccessors {
         return this;
     }
 
+    /**
+     * Write into this buffer, all the bytes from the given byte array.
+     * This updates the {@linkplain #writerOffset() write offset} of this buffer by the length of the array.
+     *
+     * @param source The byte array to read from.
+     * @return This buffer.
+     */
+    default Buffer writeBytes(byte[] source) {
+        int size = source.length;
+        int woff = writerOffset();
+        writerOffset(woff + size);
+        for (int i = 0; i < size; i++) {
+            setByte(woff + i, source[i]);
+        }
+        return this;
+    }
+
     /**
      * Resets the {@linkplain #readerOffset() read offset} and the {@linkplain #writerOffset() write offset} on this
      * buffer to their initial values.

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

@@ -18,6 +18,7 @@ package io.netty.buffer.api;
 import io.netty.buffer.api.internal.Statics;
 
 import java.nio.ByteOrder;
+import java.util.function.Supplier;
 
 /**
  * Interface for {@link Buffer} allocators.
@@ -68,6 +69,38 @@ public interface BufferAllocator extends AutoCloseable {
         return allocate(size).order(order);
     }
 
+    /**
+     * Create a supplier of "constant" {@linkplain Buffer Buffers} from this allocator, that all have the given
+     * byte contents. The buffer has the same capacity as the byte array length, and its write offset is placed at the
+     * end, and its read offset is at the beginning, such that the entire buffer contents are readable.
+     * <p>
+     * The buffers produced by the supplier will have {@linkplain Buffer#isOwned() ownership}, and closing them will
+     * make them {@linkplain Buffer#isAccessible() inaccessible}, just like a normally allocated buffer.
+     * <p>
+     * The buffers produced are only "constants" in so far as they are {@linkplain Buffer#readOnly() read-only}.
+     * However, since all buffers are meant to behave the same, it is possible to make the returned buffers writeable
+     * again. Doing so will only impact the particular buffer instance, such that changing its contents will not impact
+     * any other buffer produced by the supplier.
+     * <p>
+     * It can generally be expected, but is not guaranteed, that the returned supplier is more resource efficient than
+     * allocating and copying memory with other available APIs.
+     * <p>
+     * The primary use case for this API, is when you need to repeatedly produce buffers with the same contents, and
+     * you perhaps wish to keep a {@code static final} field with these contents. This use case has previously been
+     * solved by allocating a read-only buffer with the given contents, and then slicing or duplicating it on every use.
+     * This approach had several problems. For instance, if you forget to slice, the offsets of the buffer can change
+     * in unexpected ways, since the same buffer instance is shared and accessed from many places. The buffer could also
+     * be deallocated, making the data inaccessible. Lastly, the read-only state could be changed, allowing the
+     * supposedly constant buffer to change its contents. The supplier-based API solves all of these problems, by
+     * enforcing that each usage get their own distinct buffer instance.
+     *
+     * @param bytes The byte contents of the buffers produced by the returned supplier.
+     * @return A supplier of read-only buffers with the given contents.
+     */
+    default Supplier<Buffer> constBufferSupplier(byte[] bytes) {
+        return () -> allocate(bytes.length).writeBytes(bytes).readOnly(true);
+    }
+
     /**
      * Close this allocator, freeing all of its internal resources. It is not specified if the allocator can still be
      * used after this method has been called on it.

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

@@ -15,9 +15,6 @@
  */
 package io.netty.buffer.api;
 
-import io.netty.buffer.api.memseg.HeapMemorySegmentManager;
-import io.netty.buffer.api.memseg.NativeMemorySegmentManager;
-
 import java.lang.ref.Cleaner;
 
 public interface MemoryManager {

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

@@ -356,4 +356,17 @@ public class BufferBulkAccessTest extends BufferTestSupport {
             assertEquals(source, readableSlice);
         }
     }
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void writeBytesMustWriteAllBytesFromByteArray(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buffer = allocator.allocate(8)) {
+            buffer.writeByte((byte) 1);
+            buffer.writeBytes(new byte[] {2, 3, 4, 5, 6, 7});
+            assertThat(buffer.writerOffset()).isEqualTo(7);
+            assertThat(buffer.readerOffset()).isZero();
+            assertThat(toByteArray(buffer)).containsExactly(1, 2, 3, 4, 5, 6, 7, 0);
+        }
+    }
 }

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

@@ -19,6 +19,10 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import java.util.function.Supplier;
+
+import static java.nio.ByteOrder.BIG_ENDIAN;
+import static java.nio.ByteOrder.LITTLE_ENDIAN;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -136,5 +140,29 @@ public class BufferReadOnlyTest extends BufferTestSupport {
             }
         }
     }
-    // todo read only buffer must have zero writable bytes
+
+    @ParameterizedTest
+    @MethodSource("allocators")
+    public void readOnlyBuffersCannotChangeWriteOffset(Fixture fixture) {
+        try (BufferAllocator allocator = fixture.createAllocator();
+             Buffer buf = allocator.allocate(8).readOnly(true)) {
+            assertThrows(IllegalStateException.class, () -> buf.writerOffset(4));
+        }
+    }
+
+    @Test
+    public void modifyingConstBufferDoesNotImpactSiblings() {
+        Supplier<Buffer> supplier = BufferAllocator.heap().constBufferSupplier(new byte[] {1, 2, 3, 4});
+        try (Buffer a = supplier.get();
+             Buffer b = supplier.get().order(LITTLE_ENDIAN)) {
+            a.order(BIG_ENDIAN).readOnly(false).setInt(0, 0xA1A2A3A4);
+            a.readerOffset(2);
+            assertThat(toByteArray(a)).containsExactly(0xA1, 0xA2, 0xA3, 0xA4);
+            assertThat(toByteArray(b)).containsExactly(1, 2, 3, 4);
+            assertThat(b.readerOffset()).isZero();
+            assertThat(b.order()).isEqualTo(LITTLE_ENDIAN);
+            assertThat(a.writerOffset()).isEqualTo(4);
+            assertThat(b.writerOffset()).isEqualTo(4);
+        }
+    }
 }

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

@@ -130,11 +130,23 @@ public abstract class BufferTestSupport {
     static List<Fixture> initialAllocators() {
         return List.of(
                 new Fixture("heap", BufferAllocator::heap, HEAP),
+                new Fixture("constHeap", () -> constantBufferBasedAllocator(BufferAllocator.heap()), HEAP),
+                new Fixture("constDirect", () -> constantBufferBasedAllocator(BufferAllocator.direct()),
+                        DIRECT, CLEANER),
                 new Fixture("direct", BufferAllocator::direct, DIRECT, CLEANER),
                 new Fixture("pooledHeap", BufferAllocator::pooledHeap, POOLED, HEAP),
                 new Fixture("pooledDirect", BufferAllocator::pooledDirect, POOLED, DIRECT, CLEANER));
     }
 
+    private static BufferAllocator constantBufferBasedAllocator(BufferAllocator allocator) {
+        return size -> {
+            if (size < 0) {
+                throw new IllegalArgumentException();
+            }
+            return allocator.constBufferSupplier(new byte[size]).get().readOnly(false).reset();
+        };
+    }
+
     private static Stream<Fixture> fixtureCombinations() {
         List<Fixture> initFixtures = initialAllocators();
 
@@ -234,6 +246,37 @@ public abstract class BufferTestSupport {
                         return buf;
                     }
 
+                    @Override
+                    public void close() {
+                        allocator.close();
+                    }
+                };
+            }, COMPOSITE));
+            builder.add(new Fixture(fixture + ".readOnly(true/false)", () -> {
+                var allocator = fixture.get();
+                return new BufferAllocator() {
+                    @Override
+                    public Buffer allocate(int size) {
+                        return allocator.allocate(size).readOnly(true).readOnly(false);
+                    }
+
+                    @Override
+                    public void close() {
+                        allocator.close();
+                    }
+                };
+            }, fixture.getProperties()));
+            builder.add(new Fixture(fixture + ".compose.readOnly(true/false)", () -> {
+                var allocator = fixture.get();
+                return new BufferAllocator() {
+                    @Override
+                    public Buffer allocate(int size) {
+                        try (Buffer buf = allocator.allocate(size)) {
+                            CompositeBuffer composite = CompositeBuffer.compose(allocator, buf);
+                            return composite.readOnly(true).readOnly(false);
+                        }
+                    }
+
                     @Override
                     public void close() {
                         allocator.close();
@@ -244,8 +287,7 @@ public abstract class BufferTestSupport {
 
         var stream = builder.build();
         return stream.flatMap(BufferTestSupport::injectSplits)
-                     .flatMap(BufferTestSupport::injectSlices)
-                     .flatMap(BufferTestSupport::injectReadOnlyToggling);
+                     .flatMap(BufferTestSupport::injectSlices);
     }
 
     private static Stream<Fixture> injectSplits(Fixture f) {
@@ -310,26 +352,6 @@ public abstract class BufferTestSupport {
         return builder.build();
     }
 
-    private static Stream<Fixture> injectReadOnlyToggling(Fixture f) {
-        Builder<Fixture> builder = Stream.builder();
-        builder.add(f);
-        builder.add(new Fixture(f + ".readOnly(true/false)", () -> {
-            var allocatorBase = f.get();
-            return new BufferAllocator() {
-                @Override
-                public Buffer allocate(int size) {
-                    return allocatorBase.allocate(size).readOnly(true).readOnly(false);
-                }
-
-                @Override
-                public void close() {
-                    allocatorBase.close();
-                }
-            };
-        }, f.getProperties()));
-        return builder.build();
-    }
-
     private static Properties[] concat(Properties[] props, Properties prop) {
         props = Arrays.copyOf(props, props.length + 1);
         props[props.length - 1] = prop;