Add UnsupportedMessageTypeException

- Replaced some IllegalArgumentExceptions with
  UnsupportedMessageTypeException
- MessageToMessage(Encoder|Decoder) should continue polling the
  inbound buffer if encode() or decode() returns null
  - aggregating codec can do that

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

@@ -31,7 +31,7 @@ public abstract class MessageToMessageDecoder<I, O> extends ChannelInboundHandle
                 O emsg = decode(ctx, msg);
                 if (emsg == null) {
                     // Decoder consumed a message but returned null.
-                    // Probably it needs more messages.
+                    // Probably it needs more messages because it's an aggregator.
                     continue;
                 }
 

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

@@ -31,9 +31,9 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
 
                 O emsg = encode(ctx, msg);
                 if (emsg == null) {
-                    throw new IllegalArgumentException(
-                            "encode() returned null. unsupported message type? " +
-                            msg.getClass().getName());
+                    // encode() might be waiting for more inbound messages to generate
+                    // an aggregated message - keep polling.
+                    continue;
                 }
 
                 if (unfoldAndAdd(ctx, ctx.out(), emsg)) {
@@ -103,9 +103,7 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
             }
             dst.byteBuffer().writeBytes(buf, buf.readerIndex(), buf.readableBytes());
         } else {
-            throw new IllegalArgumentException(
-                    "message cannot be written to byte buffer if it is not " +
-                    ChannelBuffer.class.getSimpleName() + '.');
+            throw new UnsupportedMessageTypeException(msg, ChannelBuffer.class);
         }
 
         return true;

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

@@ -22,8 +22,7 @@ import io.netty.buffer.ChannelBuffer;
  * operation on a {@link ChannelBuffer} in a {@link ReplayingDecoder}
  * implementation.
  */
-public class UnreplayableOperationException extends
-        UnsupportedOperationException {
+public class UnreplayableOperationException extends UnsupportedOperationException {
 
     private static final long serialVersionUID = 8577363912862364021L;
 

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

@@ -0,0 +1,50 @@
+package io.netty.handler.codec;
+
+public class UnsupportedMessageTypeException extends CodecException {
+
+    private static final long serialVersionUID = 2799598826487038726L;
+
+    public UnsupportedMessageTypeException(
+            Object message, Class<?> expectedType, Class<?>... otherExpectedTypes) {
+        super(message(
+                message == null? "null" : message.getClass().getName(),
+                expectedType, otherExpectedTypes));
+    }
+
+    public UnsupportedMessageTypeException() {
+        super();
+    }
+
+    public UnsupportedMessageTypeException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public UnsupportedMessageTypeException(String s) {
+        super(s);
+    }
+
+    public UnsupportedMessageTypeException(Throwable cause) {
+        super(cause);
+    }
+
+    private static String message(
+            String actualType, Class<?> expectedType, Class<?>... otherExpectedTypes) {
+        if (expectedType == null) {
+            throw new NullPointerException("expectedType");
+        }
+
+        StringBuilder buf = new StringBuilder(actualType);
+        buf.append(" (expected: ").append(expectedType.getName());
+
+        if (otherExpectedTypes != null) {
+            for (Class<?> t: otherExpectedTypes) {
+                if (t == null) {
+                    break;
+                }
+                buf.append(", ").append(t.getName());
+            }
+        }
+
+        return buf.append(')').toString();
+    }
+}

codec/src/main/java/io/netty/handler/codec/redis/RedisEncoder.java

@@ -19,6 +19,7 @@ import io.netty.buffer.ChannelBuffer;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelOutboundHandlerContext;
 import io.netty.handler.codec.MessageToStreamEncoder;
+import io.netty.handler.codec.UnsupportedMessageTypeException;
 
 /**
  * {@link SimpleChannelDownstreamHandler} which encodes {@link Command}'s to {@link ChannelBuffer}'s
@@ -43,7 +44,7 @@ public class RedisEncoder extends MessageToStreamEncoder<Object> {
                 }
             }
         } else {
-            throw new IllegalArgumentException("unsupported message type: " + msg.getClass().getName());
+            throw new UnsupportedMessageTypeException(msg, Command.class, Iterable.class);
         }
     }
 }