Make retained derived buffers recyclable

Related: #4333 #4421 #5128

Motivation:

slice(), duplicate() and readSlice() currently create a non-recyclable
derived buffer instance. Under heavy load, an application that creates a
lot of derived buffers can put the garbage collector under pressure.

Modifications:

- Add the following methods which creates a non-recyclable derived buffer
  - retainedSlice()
  - retainedDuplicate()
  - readRetainedSlice()
- Add the new recyclable derived buffer implementations, which has its
  own reference count value
- Add ByteBufHolder.retainedDuplicate()
- Add ByteBufHolder.replace(ByteBuf) so that..
  - a user can replace the content of the holder in a consistent way
  - copy/duplicate/retainedDuplicate() can delegate the holder
    construction to replace(ByteBuf)
- Use retainedDuplicate() and retainedSlice() wherever possible
- Miscellaneous:
  - Rename DuplicateByteBufTest to DuplicatedByteBufTest (missing 'D')
  - Make ReplayingDecoderByteBuf.reject() return an exception instead of
    throwing it so that its callers don't need to add dummy return
    statement

Result:

Derived buffers are now recycled when created via retainedSlice() and
retainedDuplicate() and derived from a pooled buffer

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

@@ -836,6 +836,13 @@ public ByteBuf readSlice(int length) {
         return slice;
     }
 
+    @Override
+    public ByteBuf readRetainedSlice(int length) {
+        ByteBuf slice = retainedSlice(readerIndex, length);
+        readerIndex += length;
+        return slice;
+    }
+
     @Override
     public ByteBuf readBytes(byte[] dst, int dstIndex, int length) {
         checkReadableBytes(length);
@@ -1159,16 +1166,31 @@ public ByteBuf duplicate() {
         return new DuplicatedAbstractByteBuf(this);
     }
 
+    @Override
+    public ByteBuf retainedDuplicate() {
+        return duplicate().retain();
+    }
+
     @Override
     public ByteBuf slice() {
         return slice(readerIndex, readableBytes());
     }
 
+    @Override
+    public ByteBuf retainedSlice() {
+        return slice().retain();
+    }
+
     @Override
     public ByteBuf slice(int index, int length) {
         return new SlicedAbstractByteBuf(this, index, length);
     }
 
+    @Override
+    public ByteBuf retainedSlice(int index, int length) {
+        return slice(index, length).retain();
+    }
+
     @Override
     public ByteBuffer nioBuffer() {
         return nioBuffer(readerIndex, readableBytes());

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

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2016 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:
+ *
+ *   http://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;
+
+import io.netty.util.Recycler.Handle;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+/**
+ * Abstract base class for derived {@link ByteBuf} implementations.
+ */
+abstract class AbstractPooledDerivedByteBuf<T> extends AbstractReferenceCountedByteBuf {
+
+    private final Handle<AbstractPooledDerivedByteBuf<T>> recyclerHandle;
+    private AbstractByteBuf buffer;
+
+    @SuppressWarnings("unchecked")
+    AbstractPooledDerivedByteBuf(Handle<? extends AbstractPooledDerivedByteBuf<T>> recyclerHandle) {
+        super(0);
+        this.recyclerHandle = (Handle<AbstractPooledDerivedByteBuf<T>>) recyclerHandle;
+    }
+
+    @Override
+    public final AbstractByteBuf unwrap() {
+        return buffer;
+    }
+
+    final <U extends AbstractPooledDerivedByteBuf<T>> U init(
+            AbstractByteBuf buffer, int readerIndex, int writerIndex, int maxCapacity) {
+
+        buffer.retain();
+        this.buffer = buffer;
+
+        boolean success = false;
+        try {
+            maxCapacity(maxCapacity);
+            setIndex(readerIndex, writerIndex);
+            setRefCnt(1);
+
+            @SuppressWarnings("unchecked")
+            final U castThis = (U) this;
+            success = true;
+            return castThis;
+        } finally {
+            if (!success) {
+                this.buffer = null;
+                buffer.release();
+            }
+        }
+    }
+
+    @Override
+    protected final void deallocate() {
+        recyclerHandle.recycle(this);
+        unwrap().release();
+    }
+
+    @Override
+    public final ByteBufAllocator alloc() {
+        return unwrap().alloc();
+    }
+
+    @Override
+    @Deprecated
+    public final ByteOrder order() {
+        return unwrap().order();
+    }
+
+    @Override
+    public boolean isReadOnly() {
+        return unwrap().isReadOnly();
+    }
+
+    @Override
+    public final boolean isDirect() {
+        return unwrap().isDirect();
+    }
+
+    @Override
+    public boolean hasArray() {
+        return unwrap().hasArray();
+    }
+
+    @Override
+    public byte[] array() {
+        return unwrap().array();
+    }
+
+    @Override
+    public boolean hasMemoryAddress() {
+        return unwrap().hasMemoryAddress();
+    }
+
+    @Override
+    public final int nioBufferCount() {
+        return unwrap().nioBufferCount();
+    }
+
+    @Override
+    public final ByteBuffer internalNioBuffer(int index, int length) {
+        return nioBuffer(index, length);
+    }
+
+    @Override
+    public final ByteBuf retainedDuplicate() {
+        return PooledDuplicatedByteBuf.newInstance(this, readerIndex(), writerIndex());
+    }
+
+    @Override
+    public final ByteBuf retainedSlice() {
+        final int index = readerIndex();
+        return retainedSlice(index, writerIndex() - index);
+    }
+
+    @Override
+    public final ByteBuf retainedSlice(int index, int length) {
+        return PooledSlicedByteBuf.newInstance(this, index, length, index);
+    }
+}

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

@@ -76,24 +76,48 @@ public ByteBuf slice() {
         return new AdvancedLeakAwareByteBuf(super.slice(), leak);
     }
 
+    @Override
+    public ByteBuf retainedSlice() {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.retainedSlice(), leak);
+    }
+
     @Override
     public ByteBuf slice(int index, int length) {
         recordLeakNonRefCountingOperation(leak);
         return new AdvancedLeakAwareByteBuf(super.slice(index, length), leak);
     }
 
+    @Override
+    public ByteBuf retainedSlice(int index, int length) {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.retainedSlice(index, length), leak);
+    }
+
     @Override
     public ByteBuf duplicate() {
         recordLeakNonRefCountingOperation(leak);
         return new AdvancedLeakAwareByteBuf(super.duplicate(), leak);
     }
 
+    @Override
+    public ByteBuf retainedDuplicate() {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.retainedDuplicate(), leak);
+    }
+
     @Override
     public ByteBuf readSlice(int length) {
         recordLeakNonRefCountingOperation(leak);
         return new AdvancedLeakAwareByteBuf(super.readSlice(length), leak);
     }
 
+    @Override
+    public ByteBuf readRetainedSlice(int length) {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.readRetainedSlice(length), leak);
+    }
+
     @Override
     public ByteBuf discardReadBytes() {
         recordLeakNonRefCountingOperation(leak);

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

@@ -58,24 +58,48 @@ public ByteBuf slice() {
         return new AdvancedLeakAwareByteBuf(super.slice(), leak);
     }
 
+    @Override
+    public ByteBuf retainedSlice() {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.retainedSlice(), leak);
+    }
+
     @Override
     public ByteBuf slice(int index, int length) {
         recordLeakNonRefCountingOperation(leak);
         return new AdvancedLeakAwareByteBuf(super.slice(index, length), leak);
     }
 
+    @Override
+    public ByteBuf retainedSlice(int index, int length) {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.retainedSlice(index, length), leak);
+    }
+
     @Override
     public ByteBuf duplicate() {
         recordLeakNonRefCountingOperation(leak);
         return new AdvancedLeakAwareByteBuf(super.duplicate(), leak);
     }
 
+    @Override
+    public ByteBuf retainedDuplicate() {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.retainedDuplicate(), leak);
+    }
+
     @Override
     public ByteBuf readSlice(int length) {
         recordLeakNonRefCountingOperation(leak);
         return new AdvancedLeakAwareByteBuf(super.readSlice(length), leak);
     }
 
+    @Override
+    public ByteBuf readRetainedSlice(int length) {
+        recordLeakNonRefCountingOperation(leak);
+        return new AdvancedLeakAwareByteBuf(super.readRetainedSlice(length), leak);
+    }
+
     @Override
     public CompositeByteBuf discardReadBytes() {
         recordLeakNonRefCountingOperation(leak);