[#882] Add a PartialFlushException which will allow to notify the user that the flush/write was only partial succesful

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

@@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelOutboundByteHandlerAdapter;
 import io.netty.channel.ChannelPromise;
+import io.netty.channel.PartialFlushException;
 
 /**
  * {@link ChannelOutboundByteHandlerAdapter} which encodes bytes in a stream-like fashion from one {@link ByteBuf} to an
@@ -49,23 +50,31 @@ public abstract class ByteToByteEncoder extends ChannelOutboundByteHandlerAdapte
     public void flush(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
         ByteBuf in = ctx.outboundByteBuffer();
         ByteBuf out = ctx.nextOutboundByteBuffer();
+        boolean encoded = false;
 
         while (in.readable()) {
             int oldInSize = in.readableBytes();
             try {
                 encode(ctx, in, out);
+                encoded = true;
             } catch (Throwable t) {
+                Throwable cause;
                 if (t instanceof CodecException) {
-                    ctx.fireExceptionCaught(t);
+                    cause = t;
                 } else {
-                    ctx.fireExceptionCaught(new EncoderException(t));
+                    cause = new EncoderException(t);
                 }
+                if (encoded) {
+                    cause = new PartialFlushException("Unable to encoded all bytes", cause);
+                }
+                in.discardSomeReadBytes();
+                promise.setFailure(cause);
+                return;
             }
             if (oldInSize == in.readableBytes()) {
                 break;
             }
         }
-
         in.discardSomeReadBytes();
         ctx.flush(promise);
     }

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

@@ -22,6 +22,7 @@ import io.netty.channel.ChannelOutboundMessageHandlerAdapter;
 import io.netty.channel.ChannelHandlerUtil;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
+import io.netty.channel.PartialFlushException;
 
 /**
  * {@link ChannelOutboundMessageHandlerAdapter} which encodes from one message to an other message
@@ -59,6 +60,8 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessa
     @Override
     public void flush(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
         MessageBuf<I> in = ctx.outboundMessageBuffer();
+        boolean encoded = false;
+
         for (;;) {
             try {
                 Object msg = in.poll();
@@ -84,6 +87,7 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessa
                     if (omsg == imsg) {
                         free = false;
                     }
+                    encoded = true;
                     ChannelHandlerUtil.unfoldAndAdd(ctx, omsg, false);
                 } finally {
                     if (free) {
@@ -91,14 +95,19 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundMessa
                     }
                 }
             } catch (Throwable t) {
+                Throwable cause;
                 if (t instanceof CodecException) {
-                    ctx.fireExceptionCaught(t);
+                    cause = t;
                 } else {
-                    ctx.fireExceptionCaught(new EncoderException(t));
+                    cause = new EncoderException(t);
+                }
+                if (encoded) {
+                    cause = new PartialFlushException("Unable to encoded all messages", cause);
+                }
+                promise.setFailure(cause);
+                return;
             }
         }
-        }
-
         ctx.flush(promise);
     }
 

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

@@ -69,6 +69,13 @@ public interface ChannelOutboundInvoker {
     /**
      * Flush all pending data which belongs to this ChannelOutboundInvoker and notify the {@link ChannelFuture}
      * once the operation completes, either because the operation was successful or because of an error.
+     * <p>
+     * Be aware that the flush could be only partially successful. In such cases the {@link ChannelFuture} will be
+     * failed with an {@link PartialFlushException}. So if you are interested to know if it was partial successful you
+     * need to check if the returned {@link ChannelFuture#cause()} returns an instance of
+     * {@link PartialFlushException}. In such cases you may want to call {@link #flush(ChannelPromise)} or
+     * {@link #flush()} to flush the rest of the data or just close the connection via {@link #close(ChannelPromise)} or
+     * {@link #close()}  if it is not possible to recover.
      */
     ChannelFuture flush();
 
@@ -76,7 +83,13 @@ public interface ChannelOutboundInvoker {
      * Write a message via this ChannelOutboundInvoker and notify the {@link ChannelFuture}
      * once the operation completes, either because the operation was successful or because of an error.
      *
-     * If you want to write a {@link FileRegion} use {@link #sendFile(FileRegion)}
+     * If you want to write a {@link FileRegion} use {@link #sendFile(FileRegion)}.
+     * <p>
+     * Be aware that the write could be only partially successful as the message may need to get encoded before write it
+     * to the remote peer. In such cases the {@link ChannelFuture} will be failed with a {@link PartialFlushException}.
+     * In such cases you may want to call {@link #flush(ChannelPromise)} or  {@link #flush()} to flush the rest of the
+     * data or just close the connection via {@link #close(ChannelPromise)} or {@link #close()} if it is not possible
+     * to recover.
      */
     ChannelFuture write(Object message);
 
@@ -152,6 +165,13 @@ public interface ChannelOutboundInvoker {
     /**
      * Flush all pending data which belongs to this ChannelOutboundInvoker and notify the {@link ChannelPromise}
      * once the operation completes, either because the operation was successful or because of an error.
+     * <p>
+     * Be aware that the flush could be only partially successful. In such cases the {@link ChannelFuture} will be
+     * failed with an {@link PartialFlushException}. So if you are interested to know if it was partial successful you
+     * need to check if the returned {@link ChannelFuture#cause()} returns an instance of
+     * {@link PartialFlushException}. In such cases you may want to call {@link #flush(ChannelPromise)} or
+     * {@link #flush()} to flush the rest of the data or just close the connection via {@link #close(ChannelPromise)} or
+     * {@link #close()}  if it is not possible to recover.
      *
      * The given {@link ChannelPromise} will be notified.
      */
@@ -163,6 +183,12 @@ public interface ChannelOutboundInvoker {
      *
      * If you want to write a {@link FileRegion} use {@link #sendFile(FileRegion)}
      * The given {@link ChannelPromise} will be notified and also returned.
+     * <p>
+     * Be aware that the write could be only partially successful as the message may need to get encoded before write it
+     * to the remote peer. In such cases the {@link ChannelFuture} will be failed with a {@link PartialFlushException}.
+     * In such cases you may want to call {@link #flush(ChannelPromise)} or  {@link #flush()} to flush the rest of the
+     * data or just close the connection via {@link #close(ChannelPromise)} or {@link #close()} if it is not possible
+     * to recover.
      */
     ChannelFuture write(Object message, ChannelPromise promise);
 

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2013 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.channel;
+
+/**
+ * Special {@link RuntimeException} which will be used by {@link ChannelOutboundInvoker#flush(ChannelPromise)},
+ * {@link ChannelOutboundInvoker#flush()}, {@link ChannelOutboundInvoker#write(Object)} and
+ * {@link ChannelOutboundInvoker#write(Object, ChannelPromise)} if the operation was only partial successful.
+ */
+public class PartialFlushException extends RuntimeException {
+    private static final long serialVersionUID = 990261865971015004L;
+
+    public PartialFlushException(String msg, Throwable cause) {
+        super(msg, cause);
+    }
+
+    public PartialFlushException(Throwable cause) {
+        super(cause);
+    }
+}