Add ABORT signal to Channel*MessageHandlerAdapter

- Related: #1030

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

@@ -18,6 +18,7 @@ package io.netty.channel;
 import io.netty.buffer.BufUtil;
 import io.netty.buffer.MessageBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.util.internal.Signal;
 import io.netty.util.internal.TypeParameterMatcher;
 
 /**
@@ -45,6 +46,11 @@ import io.netty.util.internal.TypeParameterMatcher;
 public abstract class ChannelInboundMessageHandlerAdapter<I>
         extends ChannelStateHandlerAdapter implements ChannelInboundMessageHandler<I> {
 
+    /**
+     * Thrown by {@link #messageReceived(ChannelHandlerContext, Object)} to abort message processing.
+     */
+    protected static final Signal ABORT = new Signal(ChannelInboundMessageHandlerAdapter.class.getName() + ".ABORT");
+
     private final TypeParameterMatcher msgMatcher;
 
     protected ChannelInboundMessageHandlerAdapter() {
@@ -70,11 +76,11 @@ public abstract class ChannelInboundMessageHandlerAdapter<I>
 
     @Override
     public final void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception {
-        if (!beginMessageReceived(ctx)) {
+        MessageBuf<Object> in = ctx.inboundMessageBuffer();
+        if (in.isEmpty() || !beginMessageReceived(ctx)) {
             return;
         }
 
-        MessageBuf<Object> in = ctx.inboundMessageBuffer();
         MessageBuf<Object> out = ctx.nextInboundMessageBuffer();
         int oldOutSize = out.size();
         try {
@@ -103,6 +109,9 @@ public abstract class ChannelInboundMessageHandlerAdapter<I>
                 I imsg = (I) msg;
                 try {
                     messageReceived(ctx, imsg);
+                } catch (Signal abort) {
+                    abort.expect(ABORT);
+                    break;
                 } finally {
                     freeInboundMessage(imsg);
                 }

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

@@ -19,6 +19,7 @@ import io.netty.buffer.BufUtil;
 import io.netty.buffer.MessageBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.logging.InternalLoggerFactory;
+import io.netty.util.internal.Signal;
 import io.netty.util.internal.TypeParameterMatcher;
 
 /**
@@ -29,6 +30,11 @@ import io.netty.util.internal.TypeParameterMatcher;
 public abstract class ChannelOutboundMessageHandlerAdapter<I>
         extends ChannelOperationHandlerAdapter implements ChannelOutboundMessageHandler<I> {
 
+    /**
+     * Thrown by {@link #flush(ChannelHandlerContext, Object)} to abort message processing.
+     */
+    protected static final Signal ABORT = new Signal(ChannelOutboundMessageHandlerAdapter.class.getName() + ".ABORT");
+
     private final TypeParameterMatcher msgMatcher;
     private boolean closeOnFailedFlush = true;
 
@@ -76,8 +82,12 @@ public abstract class ChannelOutboundMessageHandlerAdapter<I>
         MessageBuf<Object> out = null;
 
         final int inSize = in.size();
-        int processed = 0;
+        if (inSize == 0) {
+            ctx.flush(promise);
+            return;
+        }
 
+        int processed = 0;
         try {
             beginFlush(ctx);
             for (;;) {
@@ -105,8 +115,16 @@ public abstract class ChannelOutboundMessageHandlerAdapter<I>
                 }
             }
         } catch (Throwable t) {
-            fail(ctx, promise, new PartialFlushException(
+            PartialFlushException pfe;
-                    processed + " out of " + inSize + " message(s) flushed; " + in.size() + " left", t));
+            String msg = processed + " out of " + inSize + " message(s) flushed";
+            if (t instanceof Signal) {
+                Signal abort = (Signal) t;
+                abort.expect(ABORT);
+                pfe = new PartialFlushException("aborted by " + getClass().getSimpleName() + ": " + msg);
+            } else {
+                pfe = new PartialFlushException(msg, t);
+            }
+            fail(ctx, promise, pfe);
         }
 
         try {

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

@@ -23,8 +23,12 @@ package io.netty.channel;
 public class PartialFlushException extends RuntimeException {
     private static final long serialVersionUID = 990261865971015004L;
 
-    public PartialFlushException(String msg, Throwable cause) {
+    public PartialFlushException(String message) {
-        super(msg, cause);
+        super(message);
+    }
+
+    public PartialFlushException(String message, Throwable cause) {
+        super(message, cause);
     }
 
     public PartialFlushException(Throwable cause) {