Allow to specify the used buffer type for ChannelInboundByteBufHandler and ChannelOutboundByteBufHandler by configuration. As default it tries to use a direct ByteBuf

codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyOrHttpChooser.java

@@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandler;
 import io.netty.channel.ChannelInboundMessageHandler;
 import io.netty.channel.ChannelPipeline;
@@ -63,7 +64,7 @@ public abstract class SpdyOrHttpChooser extends ChannelDuplexHandler implements
 
     @Override
     public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override

codec/src/main/java/io/netty/handler/codec/FixedLengthFrameDecoder.java

@@ -17,6 +17,7 @@ package io.netty.handler.codec;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 
 /**
  * A decoder that splits the received {@link ByteBuf}s by the fixed number
@@ -67,7 +68,7 @@ public class FixedLengthFrameDecoder extends ByteToMessageDecoder {
     @Override
     public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
         if (allocateFullBuffer) {
-            return ctx.alloc().ioBuffer(frameLength);
+            return ChannelHandlerUtil.allocate(ctx, frameLength);
         } else {
             return super.newInboundBuffer(ctx);
         }

example/src/main/java/io/netty/example/udt/echo/bytes/ByteEchoClientHandler.java

@@ -20,6 +20,7 @@ import com.yammer.metrics.core.Meter;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandlerAdapter;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.udt.nio.NioUdtProvider;
@@ -76,7 +77,7 @@ public class ByteEchoClientHandler extends ChannelInboundByteHandlerAdapter {
     @Override
     public ByteBuf newInboundBuffer(final ChannelHandlerContext ctx)
             throws Exception {
-        return ctx.alloc().ioBuffer(
+        return ChannelHandlerUtil.allocate(ctx,
                 ctx.channel().config().getOption(ChannelOption.SO_RCVBUF));
     }
 

example/src/main/java/io/netty/example/udt/echo/bytes/ByteEchoServerHandler.java

@@ -18,6 +18,7 @@ package io.netty.example.udt.echo.bytes;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandlerAdapter;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.udt.nio.NioUdtProvider;
@@ -57,7 +58,7 @@ public class ByteEchoServerHandler extends ChannelInboundByteHandlerAdapter {
     @Override
     public ByteBuf newInboundBuffer(final ChannelHandlerContext ctx)
             throws Exception {
-        return ctx.alloc().ioBuffer(
+        return ChannelHandlerUtil.allocate(ctx,
                 ctx.channel().config().getOption(ChannelOption.SO_RCVBUF));
     }
 

handler/src/main/java/io/netty/handler/logging/ByteLoggingHandler.java

@@ -17,6 +17,7 @@ package io.netty.handler.logging;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandler;
 import io.netty.channel.ChannelOutboundByteHandler;
 import io.netty.channel.ChannelPromise;
@@ -108,7 +109,7 @@ public class ByteLoggingHandler
 
     @Override
     public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override
@@ -123,7 +124,7 @@ public class ByteLoggingHandler
 
     @Override
     public ByteBuf newOutboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override

handler/src/main/java/io/netty/handler/ssl/SslHandler.java

@@ -23,6 +23,7 @@ import io.netty.channel.ChannelFlushPromiseNotifier;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandler;
 import io.netty.channel.ChannelOutboundByteHandler;
 import io.netty.channel.ChannelPipeline;
@@ -380,7 +381,7 @@ public class SslHandler
 
     @Override
     public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override
@@ -395,7 +396,7 @@ public class SslHandler
 
     @Override
     public ByteBuf newOutboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override

testsuite/src/test/java/io/netty/testsuite/transport/sctp/SctpEchoTest.java

@@ -21,6 +21,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.sctp.SctpChannel;
@@ -160,7 +161,7 @@ public class SctpEchoTest extends AbstractSctpTest {
 
         @Override
         public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
-            return ctx.alloc().ioBuffer(0, maxInboundBufferSize);
+            return ChannelHandlerUtil.allocate(ctx, 0, maxInboundBufferSize);
         }
 
         @Override

testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketEchoTest.java

@@ -21,6 +21,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandlerAdapter;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
@@ -186,7 +187,7 @@ public class SocketEchoTest extends AbstractSocketTest {
 
         @Override
         public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
-            return ctx.alloc().ioBuffer(0, maxInboundBufferSize);
+            return ChannelHandlerUtil.allocate(ctx, 0, maxInboundBufferSize);
         }
 
         @Override

transport-rxtx/src/main/java/io/netty/channel/rxtx/DefaultRxtxChannelConfig.java

@@ -195,4 +195,9 @@ final class DefaultRxtxChannelConfig extends DefaultChannelConfig implements Rxt
     public RxtxChannelConfig setAutoRead(boolean autoRead) {
         return (RxtxChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public RxtxChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (RxtxChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannelConfig.java

@@ -269,4 +269,7 @@ public interface RxtxChannelConfig extends ChannelConfig {
 
     @Override
     RxtxChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    RxtxChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpChannelConfig.java

@@ -19,6 +19,7 @@ package io.netty.channel.sctp;
 import com.sun.nio.sctp.SctpChannel;
 import com.sun.nio.sctp.SctpStandardSocketOptions;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -189,4 +190,9 @@ public class DefaultSctpChannelConfig extends DefaultChannelConfig implements Sc
     public SctpChannelConfig setAutoRead(boolean autoRead) {
         return (SctpChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public SctpChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (SctpChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport-sctp/src/main/java/io/netty/channel/sctp/DefaultSctpServerChannelConfig.java

@@ -173,4 +173,9 @@ public class DefaultSctpServerChannelConfig extends DefaultChannelConfig impleme
     public SctpServerChannelConfig setAutoRead(boolean autoRead) {
         return (SctpServerChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public SctpServerChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (SctpServerChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport-sctp/src/main/java/io/netty/channel/sctp/SctpChannelConfig.java

@@ -107,4 +107,7 @@ public interface SctpChannelConfig extends ChannelConfig {
 
     @Override
     SctpChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    SctpChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport-sctp/src/main/java/io/netty/channel/sctp/SctpServerChannelConfig.java

@@ -102,4 +102,7 @@ public interface SctpServerChannelConfig extends ChannelConfig {
 
     @Override
     SctpServerChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    SctpServerChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport-udt/src/main/java/io/netty/channel/udt/DefaultUdtChannelConfig.java

@@ -19,6 +19,7 @@ import com.barchart.udt.OptionUDT;
 import com.barchart.udt.SocketUDT;
 import com.barchart.udt.nio.ChannelUDT;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
 
@@ -254,4 +255,10 @@ public class DefaultUdtChannelConfig extends DefaultChannelConfig implements
         super.setAutoRead(autoRead);
         return this;
     }
+
+    @Override
+    public UdtChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        super.setDefaultHandlerByteBufType(type);
+        return this;
+    }
 }

transport-udt/src/main/java/io/netty/channel/udt/DefaultUdtServerChannelConfig.java

@@ -157,4 +157,10 @@ public class DefaultUdtServerChannelConfig extends DefaultUdtChannelConfig
         super.setAutoRead(autoRead);
         return this;
     }
+
+    @Override
+    public UdtServerChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        super.setDefaultHandlerByteBufType(type);
+        return this;
+    }
 }

transport-udt/src/main/java/io/netty/channel/udt/UdtChannelConfig.java

@@ -162,4 +162,6 @@ public interface UdtChannelConfig extends ChannelConfig {
      */
     UdtChannelConfig setSystemSendBufferSize(int size);
 
+    @Override
+    UdtChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport-udt/src/main/java/io/netty/channel/udt/UdtServerChannelConfig.java

@@ -80,4 +80,6 @@ public interface UdtServerChannelConfig extends UdtChannelConfig {
     @Override
     UdtServerChannelConfig setSystemSendBufferSize(int size);
 
+    @Override
+    UdtServerChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport-udt/src/test/java/io/netty/test/udt/util/EchoByteHandler.java

@@ -20,6 +20,7 @@ import com.yammer.metrics.core.Meter;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandlerAdapter;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.udt.nio.NioUdtProvider;
@@ -97,7 +98,7 @@ public class EchoByteHandler extends ChannelInboundByteHandlerAdapter {
 
     @Override
     public ByteBuf newInboundBuffer(final ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer(
+        return ChannelHandlerUtil.allocate(ctx,
                 ctx.channel().config().getOption(ChannelOption.SO_RCVBUF));
     }
 }

transport/src/main/java/io/netty/channel/ChannelConfig.java

@@ -15,6 +15,7 @@
  */
 package io.netty.channel;
 
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.socket.SocketChannelConfig;
 
@@ -62,6 +63,12 @@ import java.util.Map;
  */
 public interface ChannelConfig {
 
+    enum ChannelHandlerByteBufType {
+        HEAP,
+        DIRECT,
+        PREFER_DIRECT
+    }
+
     /**
      * Return all set {@link ChannelOption}'s.
      */
@@ -162,4 +169,23 @@ public interface ChannelConfig {
      * need to call it at all. The default value is {@code true}.
      */
     ChannelConfig setAutoRead(boolean autoRead);
+
+    /**
+     * Returns the {@link ChannelHandlerByteBufType} which is used to determine what kind of {@link ByteBuf} will
+     * be created by the {@link ChannelInboundByteHandler#newInboundBuffer(ChannelHandlerContext)} and
+     * {@link ChannelOutboundByteHandler#newOutboundBuffer(ChannelHandlerContext)} methods.
+     * <p>
+     * The implementation of {@link ChannelInboundByteHandler} or {@link ChannelOutboundByteHandler} may still return
+     * another {@link ByteBuf} if it depends on a special type.
+     *
+     * The default is {@link ChannelHandlerByteBufType#PREFER_DIRECT}.
+     */
+    ChannelHandlerByteBufType getDefaultHandlerByteBufType();
+
+    /**
+     * Sets the {@link ChannelHandlerByteBufType} which is used to determine what kind of {@link ByteBuf} will
+     * be created by the {@link ChannelInboundByteHandler#newInboundBuffer(ChannelHandlerContext)} and
+     * {@link ChannelOutboundByteHandler#newOutboundBuffer(ChannelHandlerContext)} methods.
+     */
+    ChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/main/java/io/netty/channel/ChannelHandlerUtil.java

@@ -17,6 +17,7 @@
 package io.netty.channel;
 
 import io.netty.buffer.BufUtil;
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.MessageBuf;
 import io.netty.util.Signal;
 import io.netty.util.internal.logging.InternalLogger;
@@ -151,6 +152,56 @@ public final class ChannelHandlerUtil {
         }
     }
 
+    /**
+     * Allocate a {@link ByteBuf} taking the {@link ChannelConfig#getDefaultHandlerByteBufType()}
+     * setting into account.
+     */
+    public static ByteBuf allocate(ChannelHandlerContext ctx) {
+        switch(ctx.channel().config().getDefaultHandlerByteBufType()) {
+            case DIRECT:
+                return ctx.alloc().directBuffer();
+            case PREFER_DIRECT:
+                return ctx.alloc().ioBuffer();
+            case HEAP:
+                return ctx.alloc().heapBuffer();
+            default:
+                throw new IllegalStateException();
+        }
+    }
+
+    /**
+     * Allocate a {@link ByteBuf} taking the {@link ChannelConfig#getDefaultHandlerByteBufType()}
+     * setting into account.
+     */
+    public static ByteBuf allocate(ChannelHandlerContext ctx, int initialCapacity) {
+        switch(ctx.channel().config().getDefaultHandlerByteBufType()) {
+            case DIRECT:
+                return ctx.alloc().directBuffer(initialCapacity);
+            case PREFER_DIRECT:
+                return ctx.alloc().ioBuffer(initialCapacity);
+            case HEAP:
+                return ctx.alloc().heapBuffer(initialCapacity);
+            default:
+                throw new IllegalStateException();
+        }
+    }
+
+    /**
+     * Allocate a {@link ByteBuf} taking the {@link ChannelConfig#getDefaultHandlerByteBufType()}
+     * setting into account.
+     */
+    public static ByteBuf allocate(ChannelHandlerContext ctx, int initialCapacity, int maxCapacity) {
+        switch(ctx.channel().config().getDefaultHandlerByteBufType()) {
+            case DIRECT:
+                return ctx.alloc().directBuffer(initialCapacity, maxCapacity);
+            case PREFER_DIRECT:
+                return ctx.alloc().ioBuffer(initialCapacity, maxCapacity);
+            case HEAP:
+                return ctx.alloc().heapBuffer(initialCapacity, maxCapacity);
+            default:
+                throw new IllegalStateException();
+        }
+    }
     private ChannelHandlerUtil() { }
 
     public interface SingleInboundMessageHandler<T> {
@@ -162,10 +213,10 @@ public final class ChannelHandlerUtil {
         boolean acceptInboundMessage(Object msg) throws Exception;
 
         /**
-         * Will get notified once {@link #inboundBufferUpdated(ChannelHandlerContext)} was called.
+         * Will get notified once {@link ChannelStateHandler#inboundBufferUpdated(ChannelHandlerContext)} was called.
          *
          * If this method returns {@code false} no further processing of the {@link MessageBuf}
-         * will be done until the next call of {@link #inboundBufferUpdated(ChannelHandlerContext)}.
+         * will be done until the next call of {@link ChannelStateHandler#inboundBufferUpdated(ChannelHandlerContext)}.
          *
          * This will return {@code true} by default, and may get overriden by sub-classes for
          * special handling.
@@ -201,7 +252,8 @@ public final class ChannelHandlerUtil {
         boolean acceptOutboundMessage(Object msg) throws Exception;
 
         /**
-         * Will get notified once {@link #flush(ChannelHandlerContext, ChannelPromise)} was called.
+         * Will get notified once {@link ChannelOperationHandler#flush(ChannelHandlerContext, ChannelPromise)}
+         * was called.
          *
          * @param ctx           the {@link ChannelHandlerContext} which this {@link ChannelHandler} belongs to
          */
@@ -216,7 +268,7 @@ public final class ChannelHandlerUtil {
         void flush(ChannelHandlerContext ctx, T msg) throws Exception;
 
         /**
-         * Is called when {@link #flush(ChannelHandlerContext, ChannelPromise)} returns.
+         * Is called when {@link ChannelOperationHandler#flush(ChannelHandlerContext, ChannelPromise)} returns.
          *
          * Super-classes may-override this for special handling.
          *

transport/src/main/java/io/netty/channel/ChannelInboundByteHandler.java

@@ -22,6 +22,12 @@ import io.netty.buffer.ByteBuf;
  *
  */
 public interface ChannelInboundByteHandler extends ChannelInboundHandler {
+    /**
+     * Return the {@link ByteBuf} which will be used for inbound data for the given {@link ChannelHandlerContext}.
+     * Implementations should take {@link ChannelConfig#getDefaultHandlerByteBufType()} into account.
+     * <p>
+     * Use of {@link ChannelHandlerUtil#allocate(ChannelHandlerContext)} is adviced.
+     */
     @Override
     ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception;
 

transport/src/main/java/io/netty/channel/ChannelInboundByteHandlerAdapter.java

@@ -33,7 +33,7 @@ public abstract class ChannelInboundByteHandlerAdapter
      */
     @Override
     public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override

transport/src/main/java/io/netty/channel/ChannelOption.java

@@ -44,7 +44,8 @@ public class ChannelOption<T> extends UniqueName {
             new ChannelOption<Boolean>("ALLOW_HALF_CLOSURE");
     public static final ChannelOption<Boolean> AUTO_READ =
             new ChannelOption<Boolean>("AUTO_READ");
-
+    public static final ChannelOption<ChannelConfig.ChannelHandlerByteBufType> DEFAULT_HANDLER_BYTEBUF_TYPE =
+            new ChannelOption<ChannelConfig.ChannelHandlerByteBufType>("DEFAULT_HANDLER_BYTEBUF_TYPE");
     public static final ChannelOption<Boolean> SO_BROADCAST =
             new ChannelOption<Boolean>("SO_BROADCAST");
     public static final ChannelOption<Boolean> SO_KEEPALIVE =

transport/src/main/java/io/netty/channel/ChannelOutboundByteHandler.java

@@ -21,6 +21,12 @@ import io.netty.buffer.ByteBuf;
  * {@link ChannelOutboundHandler} which operates on bytes which are hold in a {@link ByteBuf}.
  */
 public interface ChannelOutboundByteHandler extends ChannelOutboundHandler {
+    /**
+     * Return the {@link ByteBuf} which will be used for outbound data for the given {@link ChannelHandlerContext}.
+     * Implementations should take {@link ChannelConfig#getDefaultHandlerByteBufType()} into account.
+     * <p>
+     * Use of {@link ChannelHandlerUtil#allocate(ChannelHandlerContext)} is adviced.
+     */
     @Override
     ByteBuf newOutboundBuffer(ChannelHandlerContext ctx) throws Exception;
 

transport/src/main/java/io/netty/channel/ChannelOutboundByteHandlerAdapter.java

@@ -24,7 +24,7 @@ public abstract class ChannelOutboundByteHandlerAdapter
         extends ChannelOperationHandlerAdapter implements ChannelOutboundByteHandler {
     @Override
     public ByteBuf newOutboundBuffer(ChannelHandlerContext ctx) throws Exception {
-        return ctx.alloc().ioBuffer();
+        return ChannelHandlerUtil.allocate(ctx);
     }
 
     @Override

transport/src/main/java/io/netty/channel/DefaultChannelConfig.java

@@ -35,6 +35,7 @@ public class DefaultChannelConfig implements ChannelConfig {
 
     protected final Channel channel;
 
+    private volatile ChannelHandlerByteBufType handlerByteBufType = ChannelHandlerByteBufType.PREFER_DIRECT;
     private volatile ByteBufAllocator allocator = DEFAULT_ALLOCATOR;
     private volatile int connectTimeoutMillis = DEFAULT_CONNECT_TIMEOUT;
     private volatile int writeSpinCount = 16;
@@ -49,7 +50,8 @@ public class DefaultChannelConfig implements ChannelConfig {
 
     @Override
     public Map<ChannelOption<?>, Object> getOptions() {
-        return getOptions(null, CONNECT_TIMEOUT_MILLIS, WRITE_SPIN_COUNT, ALLOCATOR, AUTO_READ);
+        return getOptions(null, CONNECT_TIMEOUT_MILLIS, WRITE_SPIN_COUNT, ALLOCATOR, AUTO_READ,
+                DEFAULT_HANDLER_BYTEBUF_TYPE);
     }
 
     protected Map<ChannelOption<?>, Object> getOptions(
@@ -99,6 +101,9 @@ public class DefaultChannelConfig implements ChannelConfig {
         if (option == AUTO_READ) {
             return (T) Boolean.valueOf(isAutoRead());
         }
+        if (option == DEFAULT_ALLOCATOR) {
+            return (T) getDefaultHandlerByteBufType();
+        }
 
         return null;
     }
@@ -115,6 +120,8 @@ public class DefaultChannelConfig implements ChannelConfig {
             setAllocator((ByteBufAllocator) value);
         } else if (option == AUTO_READ) {
             setAutoRead((Boolean) value);
+        } else if (option == DEFAULT_HANDLER_BYTEBUF_TYPE) {
+            setDefaultHandlerByteBufType((ChannelHandlerByteBufType) value);
         } else {
             return false;
         }
@@ -187,4 +194,15 @@ public class DefaultChannelConfig implements ChannelConfig {
         }
         return this;
     }
+
+    @Override
+    public ChannelHandlerByteBufType getDefaultHandlerByteBufType() {
+        return handlerByteBufType;
+    }
+
+    @Override
+    public ChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType handlerByteBufType) {
+        this.handlerByteBufType = handlerByteBufType;
+        return this;
+    }
 }

transport/src/main/java/io/netty/channel/socket/DatagramChannelConfig.java

@@ -179,4 +179,7 @@ public interface DatagramChannelConfig extends ChannelConfig {
 
     @Override
     DatagramChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    DatagramChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/main/java/io/netty/channel/socket/DefaultDatagramChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -384,4 +385,9 @@ public class DefaultDatagramChannelConfig extends DefaultChannelConfig implement
     public DatagramChannelConfig setAutoRead(boolean autoRead) {
         return (DatagramChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public DatagramChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (DatagramChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport/src/main/java/io/netty/channel/socket/DefaultServerSocketChannelConfig.java

@@ -162,4 +162,9 @@ public class DefaultServerSocketChannelConfig extends DefaultChannelConfig
     public ServerSocketChannelConfig setAutoRead(boolean autoRead) {
         return (ServerSocketChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public ServerSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (ServerSocketChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport/src/main/java/io/netty/channel/socket/DefaultSocketChannelConfig.java

@@ -16,6 +16,7 @@
 package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.DefaultChannelConfig;
@@ -296,4 +297,9 @@ public class DefaultSocketChannelConfig extends DefaultChannelConfig
     public SocketChannelConfig setAutoRead(boolean autoRead) {
         return (SocketChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public SocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (SocketChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport/src/main/java/io/netty/channel/socket/ServerSocketChannelConfig.java

@@ -93,4 +93,7 @@ public interface ServerSocketChannelConfig extends ChannelConfig {
 
     @Override
     ServerSocketChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    ServerSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/main/java/io/netty/channel/socket/SocketChannelConfig.java

@@ -17,9 +17,9 @@ package io.netty.channel.socket;
 
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.ChannelConfig;
-import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelOption;
+import io.netty.channel.ChannelStateHandler;
 
 import java.net.Socket;
 import java.net.StandardSocketOptions;
@@ -148,7 +148,8 @@ public interface SocketChannelConfig extends ChannelConfig {
     /**
      * Sets whether the channel should not close itself when its remote peer shuts down output to
      * make the connection half-closed.  If {@code true} the connection is not closed when the
-     * remote peer shuts down output. Instead, {@link ChannelHandler#userEventTriggered(ChannelHandlerContext, Object)}
+     * remote peer shuts down output. Instead,
+     * {@link ChannelStateHandler#userEventTriggered(ChannelHandlerContext, Object)}
      * is invoked with a {@link ChannelInputShutdownEvent} object. If {@code false}, the connection
      * is closed automatically.
      */
@@ -165,4 +166,7 @@ public interface SocketChannelConfig extends ChannelConfig {
 
     @Override
     SocketChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    SocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/main/java/io/netty/channel/socket/aio/AioServerSocketChannelConfig.java

@@ -223,4 +223,10 @@ final class AioServerSocketChannelConfig extends DefaultChannelConfig implements
         super.setAutoRead(autoRead);
         return this;
     }
+
+    @Override
+    public ServerSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        super.setDefaultHandlerByteBufType(type);
+        return this;
+    }
 }

transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannelConfig.java

@@ -111,4 +111,7 @@ public interface AioSocketChannelConfig extends SocketChannelConfig {
 
     @Override
     AioSocketChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    AioSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/main/java/io/netty/channel/socket/aio/DefaultAioSocketChannelConfig.java

@@ -356,4 +356,9 @@ final class DefaultAioSocketChannelConfig extends DefaultChannelConfig
     public AioSocketChannelConfig setAutoRead(boolean autoRead) {
         return (AioSocketChannelConfig) super.setAutoRead(autoRead);
     }
+
+    @Override
+    public AioSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        return (AioSocketChannelConfig) super.setDefaultHandlerByteBufType(type);
+    }
 }

transport/src/main/java/io/netty/channel/socket/oio/DefaultOioServerSocketChannelConfig.java

@@ -130,4 +130,10 @@ public class DefaultOioServerSocketChannelConfig extends DefaultServerSocketChan
         super.setAutoRead(autoRead);
         return this;
     }
+
+    @Override
+    public OioServerSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        super.setDefaultHandlerByteBufType(type);
+        return this;
+    }
 }

transport/src/main/java/io/netty/channel/socket/oio/DefaultOioSocketChannelConfig.java

@@ -158,4 +158,10 @@ public class DefaultOioSocketChannelConfig extends DefaultSocketChannelConfig im
         super.setAutoRead(autoRead);
         return this;
     }
+
+    @Override
+    public OioSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type) {
+        super.setDefaultHandlerByteBufType(type);
+        return this;
+    }
 }

transport/src/main/java/io/netty/channel/socket/oio/OioServerSocketChannelConfig.java

@@ -72,4 +72,7 @@ public interface OioServerSocketChannelConfig extends ServerSocketChannelConfig
 
     @Override
     OioServerSocketChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    OioServerSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/main/java/io/netty/channel/socket/oio/OioSocketChannelConfig.java

@@ -87,4 +87,7 @@ public interface OioSocketChannelConfig extends SocketChannelConfig {
 
     @Override
     OioSocketChannelConfig setAutoRead(boolean autoRead);
+
+    @Override
+    OioSocketChannelConfig setDefaultHandlerByteBufType(ChannelHandlerByteBufType type);
 }

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java

@@ -22,6 +22,7 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelInboundByteHandler;
 import io.netty.channel.ChannelInboundMessageHandler;
 import io.netty.channel.ChannelInboundMessageHandlerAdapter;
@@ -413,7 +414,7 @@ public class LocalTransportThreadModelTest {
 
         @Override
         public ByteBuf newOutboundBuffer(ChannelHandlerContext ctx) throws Exception {
-            return ctx.alloc().ioBuffer();
+            return ChannelHandlerUtil.allocate(ctx);
         }
 
         @Override