From b10cf29393d9ab4af8da3ee5d90ebe3eaaafc11c Mon Sep 17 00:00:00 2001
From: Trustin Lee <trustin@gmail.com>
Date: Tue, 29 May 2012 13:34:01 -0700
Subject: [PATCH] Add isDecodable/isEncodable() to codecs to support stacked
 codecs

---
 .../codec/http/HttpChunkAggregator.java       |  4 +++
 .../codec/http/HttpContentDecoder.java        |  5 +++
 .../codec/http/HttpContentEncoder.java        | 23 ++++++++------
 .../codec/http/HttpMessageEncoder.java        |  5 +++
 .../websocketx/WebSocket00FrameEncoder.java   |  5 +++
 .../websocketx/WebSocket08FrameEncoder.java   |  5 +++
 .../handler/codec/spdy/SpdyFrameEncoder.java  | 13 ++++++++
 .../handler/codec/spdy/SpdyHttpDecoder.java   |  7 +++++
 .../handler/codec/spdy/SpdyHttpEncoder.java   |  4 +++
 .../handler/codec/LengthFieldPrepender.java   |  5 +++
 .../handler/codec/MessageToMessageCodec.java  | 28 +++++++++++++++++
 .../codec/MessageToMessageDecoder.java        | 30 +++++++++++++-----
 .../codec/MessageToMessageEncoder.java        | 31 ++++++++++++++-----
 .../handler/codec/MessageToStreamEncoder.java | 24 ++++++++++++--
 .../handler/codec/base64/Base64Decoder.java   |  5 +++
 .../handler/codec/base64/Base64Encoder.java   |  5 +++
 .../handler/codec/bytes/ByteArrayDecoder.java |  5 +++
 .../handler/codec/bytes/ByteArrayEncoder.java |  5 +++
 .../codec/protobuf/ProtobufDecoder.java       |  5 +++
 .../codec/protobuf/ProtobufEncoder.java       |  5 +++
 .../ProtobufVarint32LengthFieldPrepender.java |  5 +++
 .../CompatibleObjectEncoder.java              |  6 ++++
 .../codec/serialization/ObjectEncoder.java    | 30 +++---------------
 .../handler/codec/string/StringDecoder.java   |  5 +++
 .../handler/codec/string/StringEncoder.java   |  5 +++
 .../codec/bytes/ByteArrayDecoderTest.java     |  9 +-----
 .../codec/bytes/ByteArrayEncoderTest.java     | 10 +-----
 27 files changed, 220 insertions(+), 69 deletions(-)

diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java
index 7acc2c2a96..796c6832a6 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpChunkAggregator.java
@@ -71,6 +71,10 @@ public class HttpChunkAggregator extends MessageToMessageDecoder<Object, HttpMes
         this.maxContentLength = maxContentLength;
     }
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof HttpMessage || msg instanceof HttpChunk;
+    }
 
     @Override
     public HttpMessage decode(ChannelInboundHandlerContext<Object> ctx, Object msg) throws Exception {
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
index 408da1423c..39e24bc2c2 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecoder.java
@@ -50,6 +50,11 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<Object,
     protected HttpContentDecoder() {
     }
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof HttpMessage || msg instanceof HttpChunk;
+    }
+
     @Override
     public Object decode(ChannelInboundHandlerContext<Object> ctx, Object msg) throws Exception {
         if (msg instanceof HttpResponse && ((HttpResponse) msg).getStatus().getCode() == 100) {
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java
index 9ea38ffdeb..f758aeccb7 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentEncoder.java
@@ -47,7 +47,7 @@ import java.util.Queue;
  * so that this handler can intercept HTTP responses before {@link HttpMessageEncoder}
  * converts them into {@link ChannelBuffer}s.
  */
-public abstract class HttpContentEncoder extends MessageToMessageCodec<Object, Object, Object, Object> {
+public abstract class HttpContentEncoder extends MessageToMessageCodec<HttpMessage, HttpMessage, Object, Object> {
 
     private final Queue<String> acceptEncodingQueue = QueueFactory.createQueue(String.class);
     private volatile EncoderEmbedder<ChannelBuffer> encoder;
@@ -59,20 +59,25 @@ public abstract class HttpContentEncoder extends MessageToMessageCodec<Object, O
     }
 
     @Override
-    public Object decode(ChannelInboundHandlerContext<Object> ctx, Object msg)
-            throws Exception {
-        if (!(msg instanceof HttpMessage)) {
-            return msg;
-        }
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof HttpMessage;
+    }
 
-        HttpMessage m = (HttpMessage) msg;
-        String acceptedEncoding = m.getHeader(HttpHeaders.Names.ACCEPT_ENCODING);
+    @Override
+    public HttpMessage decode(ChannelInboundHandlerContext<HttpMessage> ctx, HttpMessage msg)
+            throws Exception {
+        String acceptedEncoding = msg.getHeader(HttpHeaders.Names.ACCEPT_ENCODING);
         if (acceptedEncoding == null) {
             acceptedEncoding = HttpHeaders.Values.IDENTITY;
         }
         boolean offered = acceptEncodingQueue.offer(acceptedEncoding);
         assert offered;
-        return m;
+        return msg;
+    }
+
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof HttpMessage || msg instanceof HttpChunk;
     }
 
     @Override
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpMessageEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpMessageEncoder.java
index 18118264b1..3b9f48d4c5 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpMessageEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpMessageEncoder.java
@@ -55,6 +55,11 @@ public abstract class HttpMessageEncoder extends MessageToStreamEncoder<Object>
     protected HttpMessageEncoder() {
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof HttpMessage || msg instanceof HttpChunk;
+    }
+
     @Override
     public void encode(ChannelOutboundHandlerContext<Object> ctx, Object msg, ChannelBuffer out) throws Exception {
         if (msg instanceof HttpMessage) {
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket00FrameEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket00FrameEncoder.java
index 15caba3f52..c0d6e8eb32 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket00FrameEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket00FrameEncoder.java
@@ -32,6 +32,11 @@ import io.netty.handler.codec.MessageToStreamEncoder;
 @Sharable
 public class WebSocket00FrameEncoder extends MessageToStreamEncoder<WebSocketFrame> {
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof WebSocketFrame;
+    }
+
     @Override
     public void encode(
             ChannelOutboundHandlerContext<WebSocketFrame> ctx,
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket08FrameEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket08FrameEncoder.java
index d096676f07..18057d482b 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket08FrameEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocket08FrameEncoder.java
@@ -93,6 +93,11 @@ public class WebSocket08FrameEncoder extends MessageToStreamEncoder<WebSocketFra
         this.maskPayload = maskPayload;
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof WebSocketFrame;
+    }
+
     @Override
     public void encode(ChannelOutboundHandlerContext<WebSocketFrame> ctx,
             WebSocketFrame msg, ChannelBuffer out) throws Exception {
diff --git a/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyFrameEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyFrameEncoder.java
index 478a645b54..1511b88d8c 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyFrameEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyFrameEncoder.java
@@ -74,6 +74,19 @@ public class SpdyFrameEncoder extends MessageToStreamEncoder<Object> {
         }
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        // TODO: Introduce a super type.
+        return msg instanceof SpdyDataFrame ||
+               msg instanceof SpdySynStreamFrame ||
+               msg instanceof SpdySynReplyFrame ||
+               msg instanceof SpdyRstStreamFrame ||
+               msg instanceof SpdySettingsFrame ||
+               msg instanceof SpdyNoOpFrame ||
+               msg instanceof SpdyGoAwayFrame ||
+               msg instanceof SpdyHeadersFrame;
+    }
+
     @Override
     public void encode(ChannelOutboundHandlerContext<Object> ctx, Object msg,
             ChannelBuffer out) throws Exception {
diff --git a/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpDecoder.java b/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpDecoder.java
index d227c8a12b..f1cb438eb5 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpDecoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpDecoder.java
@@ -59,6 +59,13 @@ public class SpdyHttpDecoder extends MessageToMessageDecoder<Object, Object> {
         this.maxContentLength = maxContentLength;
     }
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof SpdySynStreamFrame ||
+               msg instanceof SpdySynReplyFrame ||
+               msg instanceof SpdyHeadersFrame ||
+               msg instanceof SpdyDataFrame;
+    }
 
     @Override
     public Object decode(ChannelInboundHandlerContext<Object> ctx, Object msg)
diff --git a/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java b/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java
index e63e8a5465..e6885a856f 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/spdy/SpdyHttpEncoder.java
@@ -110,6 +110,10 @@ public class SpdyHttpEncoder extends MessageToMessageEncoder<Object, Object> {
 
     private volatile int currentStreamID;
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof HttpMessage || msg instanceof HttpChunk;
+    }
 
     @Override
     public Object encode(ChannelOutboundHandlerContext<Object> ctx, Object msg)
diff --git a/codec/src/main/java/io/netty/handler/codec/LengthFieldPrepender.java b/codec/src/main/java/io/netty/handler/codec/LengthFieldPrepender.java
index 8addc5a0b7..8e07e46dcc 100644
--- a/codec/src/main/java/io/netty/handler/codec/LengthFieldPrepender.java
+++ b/codec/src/main/java/io/netty/handler/codec/LengthFieldPrepender.java
@@ -96,6 +96,11 @@ public class LengthFieldPrepender extends MessageToStreamEncoder<ChannelBuffer>
         this.lengthIncludesLengthFieldLength = lengthIncludesLengthFieldLength;
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public void encode(
             ChannelOutboundHandlerContext<ChannelBuffer> ctx,
diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java b/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java
index 05745d563b..93fa073737 100644
--- a/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java
@@ -11,6 +11,11 @@ public abstract class MessageToMessageCodec<INBOUND_IN, INBOUND_OUT, OUTBOUND_IN
 
     private final MessageToMessageEncoder<OUTBOUND_IN, OUTBOUND_OUT> encoder =
             new MessageToMessageEncoder<OUTBOUND_IN, OUTBOUND_OUT>() {
+        @Override
+        public boolean isEncodable(Object msg) throws Exception {
+            return MessageToMessageCodec.this.isEncodable(msg);
+        }
+
         @Override
         public OUTBOUND_OUT encode(ChannelOutboundHandlerContext<OUTBOUND_IN> ctx, OUTBOUND_IN msg) throws Exception {
             return MessageToMessageCodec.this.encode(ctx, msg);
@@ -19,6 +24,11 @@ public abstract class MessageToMessageCodec<INBOUND_IN, INBOUND_OUT, OUTBOUND_IN
 
     private final MessageToMessageDecoder<INBOUND_IN, INBOUND_OUT> decoder =
             new MessageToMessageDecoder<INBOUND_IN, INBOUND_OUT>() {
+        @Override
+        public boolean isDecodable(Object msg) throws Exception {
+            return MessageToMessageCodec.this.isDecodable(msg);
+        }
+
         @Override
         public INBOUND_OUT decode(ChannelInboundHandlerContext<INBOUND_IN> ctx, INBOUND_IN msg) throws Exception {
             return MessageToMessageCodec.this.decode(ctx, msg);
@@ -46,6 +56,24 @@ public abstract class MessageToMessageCodec<INBOUND_IN, INBOUND_OUT, OUTBOUND_IN
         encoder.flush(ctx, future);
     }
 
+    /**
+     * Returns {@code true} if and only if the specified message can be decoded by this codec.
+     *
+     * @param msg the message
+     */
+    public boolean isDecodable(Object msg) throws Exception {
+        return true;
+    }
+
+    /**
+     * Returns {@code true} if and only if the specified message can be encoded by this codec.
+     *
+     * @param msg the message
+     */
+    public boolean isEncodable(Object msg) throws Exception {
+        return true;
+    }
+
     public abstract OUTBOUND_OUT encode(ChannelOutboundHandlerContext<OUTBOUND_IN> ctx, OUTBOUND_IN msg) throws Exception;
     public abstract INBOUND_OUT decode(ChannelInboundHandlerContext<INBOUND_IN> ctx, INBOUND_IN msg) throws Exception;
 }
diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToMessageDecoder.java b/codec/src/main/java/io/netty/handler/codec/MessageToMessageDecoder.java
index efde1a8a71..1433261f21 100644
--- a/codec/src/main/java/io/netty/handler/codec/MessageToMessageDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageToMessageDecoder.java
@@ -20,23 +20,30 @@ public abstract class MessageToMessageDecoder<I, O> extends ChannelInboundHandle
     public void inboundBufferUpdated(ChannelInboundHandlerContext<I> ctx)
             throws Exception {
         Queue<I> in = ctx.inbound().messageBuffer();
-        boolean decoded = false;
+        boolean notify = false;
         for (;;) {
             try {
-                I msg = in.poll();
+                Object msg = in.poll();
                 if (msg == null) {
                     break;
                 }
+                if (!isDecodable(msg)) {
+                    ctx.nextInboundMessageBuffer().add(msg);
+                    notify = true;
+                    continue;
+                }
 
-                O emsg = decode(ctx, msg);
-                if (emsg == null) {
+                @SuppressWarnings("unchecked")
+                I imsg = (I) msg;
+                O omsg = decode(ctx, imsg);
+                if (omsg == null) {
                     // Decoder consumed a message but returned null.
                     // Probably it needs more messages because it's an aggregator.
                     continue;
                 }
 
-                if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), emsg)) {
-                    decoded = true;
+                if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), omsg)) {
+                    notify = true;
                 }
             } catch (Throwable t) {
                 if (t instanceof CodecException) {
@@ -46,10 +53,19 @@ public abstract class MessageToMessageDecoder<I, O> extends ChannelInboundHandle
                 }
             }
         }
-        if (decoded) {
+        if (notify) {
             ctx.fireInboundBufferUpdated();
         }
     }
 
+    /**
+     * Returns {@code true} if and only if the specified message can be decoded by this decoder.
+     *
+     * @param msg the message
+     */
+    public boolean isDecodable(Object msg) throws Exception {
+        return true;
+    }
+
     public abstract O decode(ChannelInboundHandlerContext<I> ctx, I msg) throws Exception;
 }
diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java b/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java
index e8da448598..965a881dc8 100644
--- a/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java
@@ -20,23 +20,31 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
     @Override
     public void flush(ChannelOutboundHandlerContext<I> ctx, ChannelFuture future) throws Exception {
         Queue<I> in = ctx.outbound().messageBuffer();
-        boolean encoded = false;
+        boolean notify = false;
         for (;;) {
             try {
-                I msg = in.poll();
+                Object msg = in.poll();
                 if (msg == null) {
                     break;
                 }
 
-                O emsg = encode(ctx, msg);
-                if (emsg == null) {
+                if (!isEncodable(msg)) {
+                    ctx.nextOutboundMessageBuffer().add(msg);
+                    notify = true;
+                    continue;
+                }
+
+                @SuppressWarnings("unchecked")
+                I imsg = (I) msg;
+                O omsg = encode(ctx, imsg);
+                if (omsg == null) {
                     // encode() might be waiting for more inbound messages to generate
                     // an aggregated message - keep polling.
                     continue;
                 }
 
-                if (unfoldAndAdd(ctx, ctx.nextOutboundMessageBuffer(), emsg)) {
-                    encoded = true;
+                if (unfoldAndAdd(ctx, ctx.nextOutboundMessageBuffer(), omsg)) {
+                    notify = true;
                 }
             } catch (Throwable t) {
                 if (t instanceof CodecException) {
@@ -47,11 +55,20 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
             }
         }
 
-        if (encoded) {
+        if (notify) {
             ctx.flush(future);
         }
     }
 
+    /**
+     * Returns {@code true} if and only if the specified message can be encoded by this encoder.
+     *
+     * @param msg the message
+     */
+    public boolean isEncodable(Object msg) throws Exception {
+        return true;
+    }
+
     public abstract O encode(ChannelOutboundHandlerContext<I> ctx, I msg) throws Exception;
 
     static <T> boolean unfoldAndAdd(
diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToStreamEncoder.java b/codec/src/main/java/io/netty/handler/codec/MessageToStreamEncoder.java
index db2a88d607..cc3544eac7 100644
--- a/codec/src/main/java/io/netty/handler/codec/MessageToStreamEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageToStreamEncoder.java
@@ -22,15 +22,24 @@ public abstract class MessageToStreamEncoder<I> extends ChannelOutboundHandlerAd
         Queue<I> in = ctx.outbound().messageBuffer();
         ChannelBuffer out = ctx.nextOutboundByteBuffer();
 
+        boolean notify = false;
         int oldOutSize = out.readableBytes();
         for (;;) {
-            I msg = in.poll();
+            Object msg = in.poll();
             if (msg == null) {
                 break;
             }
 
+            if (!isEncodable(msg)) {
+                ctx.nextOutboundMessageBuffer().add(msg);
+                notify = true;
+                continue;
+            }
+
+            @SuppressWarnings("unchecked")
+            I imsg = (I) msg;
             try {
-                encode(ctx, msg, out);
+                encode(ctx, imsg, out);
             } catch (Throwable t) {
                 if (t instanceof CodecException) {
                     ctx.fireExceptionCaught(t);
@@ -40,10 +49,19 @@ public abstract class MessageToStreamEncoder<I> extends ChannelOutboundHandlerAd
             }
         }
 
-        if (out.readableBytes() > oldOutSize) {
+        if (out.readableBytes() > oldOutSize || notify) {
             ctx.flush(future);
         }
     }
 
+    /**
+     * Returns {@code true} if and only if the specified message can be encoded by this encoder.
+     *
+     * @param msg the message
+     */
+    public boolean isEncodable(Object msg) throws Exception {
+        return true;
+    }
+
     public abstract void encode(ChannelOutboundHandlerContext<I> ctx, I msg, ChannelBuffer out) throws Exception;
 }
diff --git a/codec/src/main/java/io/netty/handler/codec/base64/Base64Decoder.java b/codec/src/main/java/io/netty/handler/codec/base64/Base64Decoder.java
index f82ef1dfb1..9d3290ba10 100644
--- a/codec/src/main/java/io/netty/handler/codec/base64/Base64Decoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/base64/Base64Decoder.java
@@ -59,6 +59,11 @@ public class Base64Decoder extends MessageToMessageDecoder<ChannelBuffer, Channe
         this.dialect = dialect;
     }
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public ChannelBuffer decode(ChannelInboundHandlerContext<ChannelBuffer> ctx, ChannelBuffer msg) throws Exception {
         return Base64.decode(msg, msg.readerIndex(), msg.readableBytes(), dialect);
diff --git a/codec/src/main/java/io/netty/handler/codec/base64/Base64Encoder.java b/codec/src/main/java/io/netty/handler/codec/base64/Base64Encoder.java
index 13988041d6..3753a71b0a 100644
--- a/codec/src/main/java/io/netty/handler/codec/base64/Base64Encoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/base64/Base64Encoder.java
@@ -62,6 +62,11 @@ public class Base64Encoder extends MessageToMessageEncoder<ChannelBuffer, Channe
         this.dialect = dialect;
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public ChannelBuffer encode(ChannelOutboundHandlerContext<ChannelBuffer> ctx,
             ChannelBuffer msg) throws Exception {
diff --git a/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayDecoder.java b/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayDecoder.java
index 0e99f3772e..6f6c7ba84d 100644
--- a/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayDecoder.java
@@ -50,6 +50,11 @@ import io.netty.handler.codec.MessageToMessageDecoder;
  */
 public class ByteArrayDecoder extends MessageToMessageDecoder<ChannelBuffer, byte[]> {
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public byte[] decode(ChannelInboundHandlerContext<ChannelBuffer> ctx, ChannelBuffer msg) throws Exception {
         byte[] array;
diff --git a/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayEncoder.java b/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayEncoder.java
index 4300286b0a..441f4d8271 100644
--- a/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/bytes/ByteArrayEncoder.java
@@ -58,6 +58,11 @@ public class ByteArrayEncoder extends MessageToMessageEncoder<byte[], ChannelBuf
         return ChannelBufferHolders.messageBuffer();
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof byte[];
+    }
+
     @Override
     public ChannelBuffer encode(ChannelOutboundHandlerContext<byte[]> ctx, byte[] msg) throws Exception {
         if (msg.length == 0) {
diff --git a/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufDecoder.java b/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufDecoder.java
index 55359384ec..a0094305b0 100644
--- a/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufDecoder.java
@@ -82,6 +82,11 @@ public class ProtobufDecoder extends MessageToMessageDecoder<ChannelBuffer, Mess
         this.extensionRegistry = extensionRegistry;
     }
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public MessageLite decode(ChannelInboundHandlerContext<ChannelBuffer> ctx, ChannelBuffer msg) throws Exception {
         if (msg.hasArray()) {
diff --git a/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufEncoder.java b/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufEncoder.java
index ea5687285e..d348e84688 100644
--- a/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufEncoder.java
@@ -60,6 +60,11 @@ import com.google.protobuf.MessageLite;
 @Sharable
 public class ProtobufEncoder extends MessageToMessageEncoder<Object, ChannelBuffer> {
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof MessageLite || msg instanceof MessageLite.Builder;
+    }
+
     @Override
     public ChannelBuffer encode(ChannelOutboundHandlerContext<Object> ctx, Object msg) throws Exception {
         if (msg instanceof MessageLite) {
diff --git a/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufVarint32LengthFieldPrepender.java b/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufVarint32LengthFieldPrepender.java
index 1d367d42aa..064c033942 100644
--- a/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufVarint32LengthFieldPrepender.java
+++ b/codec/src/main/java/io/netty/handler/codec/protobuf/ProtobufVarint32LengthFieldPrepender.java
@@ -46,6 +46,11 @@ public class ProtobufVarint32LengthFieldPrepender extends MessageToStreamEncoder
     public ProtobufVarint32LengthFieldPrepender() {
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public void encode(
             ChannelOutboundHandlerContext<ChannelBuffer> ctx, ChannelBuffer msg, ChannelBuffer out) throws Exception {
diff --git a/codec/src/main/java/io/netty/handler/codec/serialization/CompatibleObjectEncoder.java b/codec/src/main/java/io/netty/handler/codec/serialization/CompatibleObjectEncoder.java
index e4bcf65ae3..663ed25ab3 100644
--- a/codec/src/main/java/io/netty/handler/codec/serialization/CompatibleObjectEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/serialization/CompatibleObjectEncoder.java
@@ -25,6 +25,7 @@ import io.netty.util.AttributeKey;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.OutputStream;
+import java.io.Serializable;
 
 /**
  * An encoder which serializes a Java object into a {@link ChannelBuffer}
@@ -75,6 +76,11 @@ public class CompatibleObjectEncoder extends MessageToStreamEncoder<Object> {
         return new ObjectOutputStream(out);
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof Serializable;
+    }
+
     @Override
     public void encode(ChannelOutboundHandlerContext<Object> ctx, Object msg, ChannelBuffer out) throws Exception {
         Attribute<ObjectOutputStream> oosAttr = ctx.attr(OOS);
diff --git a/codec/src/main/java/io/netty/handler/codec/serialization/ObjectEncoder.java b/codec/src/main/java/io/netty/handler/codec/serialization/ObjectEncoder.java
index 8c3515a739..4f44188e7e 100644
--- a/codec/src/main/java/io/netty/handler/codec/serialization/ObjectEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/serialization/ObjectEncoder.java
@@ -23,6 +23,7 @@ import io.netty.handler.codec.MessageToStreamEncoder;
 
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.io.Serializable;
 
 /**
  * An encoder which serializes a Java object into a {@link ChannelBuffer}.
@@ -38,32 +39,9 @@ import java.io.ObjectOutputStream;
 public class ObjectEncoder extends MessageToStreamEncoder<Object> {
     private static final byte[] LENGTH_PLACEHOLDER = new byte[4];
 
-    private final int estimatedLength;
-
-    /**
-     * Creates a new encoder with the estimated length of 512 bytes.
-     */
-    public ObjectEncoder() {
-        this(512);
-    }
-
-    /**
-     * Creates a new encoder.
-     *
-     * @param estimatedLength
-     *        the estimated byte length of the serialized form of an object.
-     *        If the length of the serialized form exceeds this value, the
-     *        internal buffer will be expanded automatically at the cost of
-     *        memory bandwidth.  If this value is too big, it will also waste
-     *        memory bandwidth.  To avoid unnecessary memory copy or allocation
-     *        cost, please specify the properly estimated value.
-     */
-    public ObjectEncoder(int estimatedLength) {
-        if (estimatedLength < 0) {
-            throw new IllegalArgumentException(
-                    "estimatedLength: " + estimatedLength);
-        }
-        this.estimatedLength = estimatedLength;
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof Serializable;
     }
 
     @Override
diff --git a/codec/src/main/java/io/netty/handler/codec/string/StringDecoder.java b/codec/src/main/java/io/netty/handler/codec/string/StringDecoder.java
index b2ab0273a2..c48865587d 100644
--- a/codec/src/main/java/io/netty/handler/codec/string/StringDecoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/string/StringDecoder.java
@@ -75,6 +75,11 @@ public class StringDecoder extends MessageToMessageDecoder<ChannelBuffer, String
         this.charset = charset;
     }
 
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        return msg instanceof ChannelBuffer;
+    }
+
     @Override
     public String decode(ChannelInboundHandlerContext<ChannelBuffer> ctx, ChannelBuffer msg) throws Exception {
         return msg.toString(charset);
diff --git a/codec/src/main/java/io/netty/handler/codec/string/StringEncoder.java b/codec/src/main/java/io/netty/handler/codec/string/StringEncoder.java
index b893e4b61c..b3052aa326 100644
--- a/codec/src/main/java/io/netty/handler/codec/string/StringEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/string/StringEncoder.java
@@ -73,6 +73,11 @@ public class StringEncoder extends MessageToMessageEncoder<String, ChannelBuffer
         this.charset = charset;
     }
 
+    @Override
+    public boolean isEncodable(Object msg) throws Exception {
+        return msg instanceof String;
+    }
+
     @Override
     public ChannelBuffer encode(ChannelOutboundHandlerContext<String> ctx, String msg) throws Exception {
         return ChannelBuffers.copiedBuffer(msg, charset);
diff --git a/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayDecoderTest.java
index ead365c5c1..2d94bad561 100644
--- a/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayDecoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayDecoderTest.java
@@ -18,7 +18,6 @@ package io.netty.handler.codec.bytes;
 import static io.netty.buffer.ChannelBuffers.*;
 import static org.hamcrest.core.Is.*;
 import static org.junit.Assert.*;
-import io.netty.handler.codec.DecoderException;
 import io.netty.handler.codec.embedder.DecoderEmbedder;
 
 import java.util.Random;
@@ -56,12 +55,6 @@ public class ByteArrayDecoderTest {
     public void testDecodeOtherType() {
         String str = "Meep!";
         embedder.offer(str);
-        try {
-            embedder.poll();
-            fail();
-        } catch (DecoderException e) {
-            // Expected
-            assertTrue(e.getCause() instanceof ClassCastException);
-        }
+        assertThat(embedder.poll(), is((Object) str));
     }
 }
diff --git a/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayEncoderTest.java b/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayEncoderTest.java
index 5365a902de..5d440fea35 100644
--- a/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayEncoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/bytes/ByteArrayEncoderTest.java
@@ -20,7 +20,6 @@ import static org.hamcrest.core.Is.*;
 import static org.hamcrest.core.IsNull.*;
 import static org.junit.Assert.*;
 import io.netty.buffer.ChannelBuffer;
-import io.netty.handler.codec.EncoderException;
 import io.netty.handler.codec.embedder.EncoderEmbedder;
 
 import java.util.Random;
@@ -58,13 +57,6 @@ public class ByteArrayEncoderTest {
     public void testEncodeOtherType() {
         String str = "Meep!";
         embedder.offer(str);
-        try {
-            embedder.poll();
-            fail();
-        } catch (EncoderException e) {
-            // Expected
-            assertTrue(e.getCause() instanceof ClassCastException);
-        }
+        assertThat(embedder.poll(), is((Object) str));
     }
-
 }