Repository: netty/netty (Java)

Issue:
ChannelOption for FIRE_EXCEPTION_ON_FAILURE?
I want to be able to react to *any* outbound channel failure.

I can add a handler to the tail of my pipeline that will call `promise.addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE)` for every outbound event, but this is not sufficient.

Specifically, in the case of a `MessageToMessageEncoder` that outputs more than one object, new promises are created for all but the last object:
https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java#L112-L127

If one of those objects then triggers a failure on a subsequent outbound handler, the exception is suppressed.  There might be other scenarios where this can happen as well.

How can I react to these failures when I don't have the opportunity to inject a `ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE`?

Would it be reasonable to have a `ChannelOption` for `FIRE_EXCEPTION_ON_FAILURE`?

### Netty version

4.1.30.Final


Hints:
@Bennett-Lynch that's a good question.. I like the idea of the `ChannelOption` but I would need to investigate if this is easily doable without breaking any API (I suspect it should be doable).
@Bennett-Lynch https://github.com/netty/netty/pull/8448 PTAL

Patch:
diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java b/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java
--- a/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java
@@ -22,6 +22,7 @@
 import io.netty.channel.ChannelPromise;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCounted;
+import io.netty.util.concurrent.PromiseCombiner;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.TypeParameterMatcher;
 
@@ -108,28 +109,36 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
             if (out != null) {
                 final int sizeMinusOne = out.size() - 1;
                 if (sizeMinusOne == 0) {
-                    ctx.write(out.get(0), promise);
+                    ctx.write(out.getUnsafe(0), promise);
                 } else if (sizeMinusOne > 0) {
                     // Check if we can use a voidPromise for our extra writes to reduce GC-Pressure
                     // See https://github.com/netty/netty/issues/2525
-                    ChannelPromise voidPromise = ctx.voidPromise();
-                    boolean isVoidPromise = promise == voidPromise;
-                    for (int i = 0; i < sizeMinusOne; i ++) {
-                        ChannelPromise p;
-                        if (isVoidPromise) {
-                            p = voidPromise;
-                        } else {
-                            p = ctx.newPromise();
-                        }
-                        ctx.write(out.getUnsafe(i), p);
+                    if (promise == ctx.voidPromise()) {
+                        writeVoidPromise(ctx, out);
+                    } else {
+                        writePromiseCombiner(ctx, out, promise);
                     }
-                    ctx.write(out.getUnsafe(sizeMinusOne), promise);
                 }
                 out.recycle();
             }
         }
     }
 
+    private static void writeVoidPromise(ChannelHandlerContext ctx, CodecOutputList out) {
+        final ChannelPromise voidPromise = ctx.voidPromise();
+        for (int i = 0; i < out.size(); i++) {
+            ctx.write(out.getUnsafe(i), voidPromise);
+        }
+    }
+
+    private static void writePromiseCombiner(ChannelHandlerContext ctx, CodecOutputList out, ChannelPromise promise) {
+        final PromiseCombiner combiner = new PromiseCombiner();
+        for (int i = 0; i < out.size(); i++) {
+            combiner.add(ctx.write(out.getUnsafe(i)));
+        }
+        combiner.finish(promise);
+    }
+
     /**
      * Encode from one message to an other. This method will be called for each written message that can be handled
      * by this encoder.


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/MessageToMessageEncoderTest.java b/codec/src/test/java/io/netty/handler/codec/MessageToMessageEncoderTest.java
--- a/codec/src/test/java/io/netty/handler/codec/MessageToMessageEncoderTest.java
+++ b/codec/src/test/java/io/netty/handler/codec/MessageToMessageEncoderTest.java
@@ -15,9 +15,14 @@
  */
 package io.netty.handler.codec;
 
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
 import io.netty.channel.embedded.EmbeddedChannel;
 import org.junit.Test;
+import static org.junit.Assert.*;
 
 import java.util.List;
 
@@ -37,4 +42,37 @@ protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) t
         });
         channel.writeOutbound(new Object());
     }
+
+    @Test
+    public void testIntermediateWriteFailures() {
+        ChannelHandler encoder = new MessageToMessageEncoder<Object>() {
+            @Override
+            protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) {
+                out.add(new Object());
+                out.add(msg);
+            }
+        };
+
+        final Exception firstWriteException = new Exception();
+
+        ChannelHandler writeThrower = new ChannelOutboundHandlerAdapter() {
+            private boolean firstWritten;
+            @Override
+            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+                if (firstWritten) {
+                    ctx.write(msg, promise);
+                } else {
+                    firstWritten = true;
+                    promise.setFailure(firstWriteException);
+                }
+            }
+        };
+
+        EmbeddedChannel channel = new EmbeddedChannel(writeThrower, encoder);
+        Object msg = new Object();
+        ChannelFuture write = channel.writeAndFlush(msg);
+        assertSame(firstWriteException, write.cause());
+        assertSame(msg, channel.readOutbound());
+        assertFalse(channel.finish());
+    }
 }