CompositeByteBuf.decompose(...) does not correctly slice content. (#8403)

Motivation:

CompositeByteBuf.decompose(...) did not correctly slice the content and so produced an incorrect representation of the data.

Modifications:

- Rewrote implementation to fix bug and also improved it to reduce GC
- Add unit tests.

Result:

Fixes https://github.com/netty/netty/issues/8400.

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

@@ -526,39 +526,31 @@ public class CompositeByteBuf extends AbstractReferenceCountedByteBuf implements
         }
 
         int componentId = toComponentIndex(offset);
-        List<ByteBuf> slice = new ArrayList<ByteBuf>(components.size());
-
+        int bytesToSlice = length;
         // The first component
         Component firstC = components.get(componentId);
-        ByteBuf first = firstC.buf.duplicate();
-        first.readerIndex(offset - firstC.offset);
+        int firstBufOffset = offset - firstC.offset;
 
-        ByteBuf buf = first;
-        int bytesToSlice = length;
-        do {
-            int readableBytes = buf.readableBytes();
-            if (bytesToSlice <= readableBytes) {
-                // Last component
-                buf.writerIndex(buf.readerIndex() + bytesToSlice);
-                slice.add(buf);
-                break;
-            } else {
-                // Not the last component
-                slice.add(buf);
-                bytesToSlice -= readableBytes;
-                componentId ++;
+        ByteBuf slice = firstC.buf.slice(firstBufOffset + firstC.buf.readerIndex(),
+                                         Math.min(firstC.length - firstBufOffset, bytesToSlice));
+        bytesToSlice -= slice.readableBytes();
 
-                // Fetch the next component.
-                buf = components.get(componentId).buf.duplicate();
+        if (bytesToSlice == 0) {
+            return Collections.singletonList(slice);
         }
+
+        List<ByteBuf> sliceList = new ArrayList<ByteBuf>(components.size() - componentId);
+        sliceList.add(slice);
+
+        // Add all the slices until there is nothing more left and then return the List.
+        do {
+            Component component = components.get(++componentId);
+            slice = component.buf.slice(component.buf.readerIndex(), Math.min(component.length, bytesToSlice));
+            bytesToSlice -= slice.readableBytes();
+            sliceList.add(slice);
         } while (bytesToSlice > 0);
 
-        // Slice all components because only readable bytes are interesting.
-        for (int i = 0; i < slice.size(); i ++) {
-            slice.set(i, slice.get(i).slice());
-        }
-
-        return slice;
+        return sliceList;
     }
 
     @Override

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

@@ -16,6 +16,7 @@
 package io.netty.buffer;
 
 import io.netty.util.ReferenceCountUtil;
+import io.netty.util.internal.PlatformDependent;
 import org.junit.Assume;
 import org.junit.Test;
 
@@ -1136,4 +1137,45 @@ public abstract class AbstractCompositeByteBufTest extends AbstractByteBufTest {
         buffer.release();
         copy.release();
     }
+
+    @Test
+    public void testDecomposeMultiple() {
+        testDecompose(150, 500, 3);
+    }
+
+    @Test
+    public void testDecomposeOne() {
+        testDecompose(310, 50, 1);
+    }
+
+    @Test
+    public void testDecomposeNone() {
+        testDecompose(310, 0, 0);
+    }
+
+    private static void testDecompose(int offset, int length, int expectedListSize) {
+        byte[] bytes = new byte[1024];
+        PlatformDependent.threadLocalRandom().nextBytes(bytes);
+        ByteBuf buf = wrappedBuffer(bytes);
+
+        CompositeByteBuf composite = compositeBuffer();
+        composite.addComponents(true,
+                                buf.retainedSlice(100, 200),
+                                buf.retainedSlice(300, 400),
+                                buf.retainedSlice(700, 100));
+
+        ByteBuf slice = composite.slice(offset, length);
+        List<ByteBuf> bufferList = composite.decompose(offset, length);
+        assertEquals(expectedListSize, bufferList.size());
+        ByteBuf wrapped = wrappedBuffer(bufferList.toArray(new ByteBuf[0]));
+
+        assertEquals(slice, wrapped);
+        composite.release();
+        buf.release();
+
+        for (ByteBuf buffer: bufferList) {
+            assertEquals(0, buffer.refCnt());
+        }
+    }
+
 }