Use the ByteBufAllocator when copy a ReadOnlyByteBufferBuf and so also be able to release it without the GC when the Cleaner is present.

Motivation:

In ReadOnlyByteBufferBuf.copy(...) we just allocated a ByteBuffer directly and wrapped it. This way it was not possible for us to free the direct memory that was used by the copy without the GC.

Modifications:

- Ensure we use the allocator when create the copy and so be able to release direct memory in a timely manner
- Add unit test
- Depending on if the to be copied buffer is direct or heap based we also allocate the same type on copy.

Result:

Fixes [#7103].

buffer/src/main/java/io/netty/buffer/ReadOnlyByteBufferBuf.java

@@ -416,11 +416,9 @@ class ReadOnlyByteBufferBuf extends AbstractReferenceCountedByteBuf {
             throw new IndexOutOfBoundsException("Too many bytes to read - Need " + (index + length));
         }
 
-        ByteBuffer dst = ByteBuffer.allocateDirect(length);
-        dst.put(src);
-        dst.order(order());
-        dst.clear();
-        return new UnpooledDirectByteBuf(alloc(), dst, maxCapacity());
+        ByteBuf dst = src.isDirect() ? alloc().directBuffer(length) : alloc().heapBuffer(length);
+        dst.writeBytes(src);
+        return dst;
     }
 
     @Override

buffer/src/test/java/io/netty/buffer/ReadOnlyByteBufferBufTest.java

@@ -15,11 +15,44 @@
  */
 package io.netty.buffer;
 
+import io.netty.util.internal.PlatformDependent;
+import org.junit.Test;
+
 import java.nio.ByteBuffer;
 
+import static org.junit.Assert.assertEquals;
+
 public class ReadOnlyByteBufferBufTest extends ReadOnlyDirectByteBufferBufTest {
     @Override
     protected ByteBuffer allocate(int size) {
         return ByteBuffer.allocate(size);
     }
+
+    @Test
+    public void testCopyDirect() {
+        testCopy(true);
+    }
+
+    @Test
+    public void testCopyHeap() {
+        testCopy(false);
+    }
+
+    private static void testCopy(boolean direct) {
+        byte[] bytes = new byte[1024];
+        PlatformDependent.threadLocalRandom().nextBytes(bytes);
+
+        ByteBuffer nioBuffer = direct ? ByteBuffer.allocateDirect(bytes.length) : ByteBuffer.allocate(bytes.length);
+        nioBuffer.put(bytes).flip();
+
+        ByteBuf buf = new ReadOnlyByteBufferBuf(UnpooledByteBufAllocator.DEFAULT, nioBuffer.asReadOnlyBuffer());
+        ByteBuf copy = buf.copy();
+
+        assertEquals(buf, copy);
+        assertEquals(buf.alloc(), copy.alloc());
+        assertEquals(buf.isDirect(), copy.isDirect());
+
+        copy.release();
+        buf.release();
+    }
 }