Do not defer closing of Channel when in flush

Motivation:

Previously, we deferred the closing of the Channel when we were flushing. This is problematic as this means that if the user adds a ChannelFutureListener, that will close the Channel, the closing will not happen until we are done with flushing. This can lead to more data is sent than expected.

Modifications:

- Do not defer closing when in flush

Result:

Correctly respect order of events and closing the Channel ASAP

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

@@ -535,24 +535,17 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
                 return;
             }
 
-            if (inFlush0) {
-                invokeLater(new OneTimeTask() {
-                    @Override
-                    public void run() {
-                        close(promise);
-                    }
-                });
-                return;
-            }
-
             if (outboundBuffer == null) {
-                // This means close() was called before so we just register a listener and return
+                // Only needed if no VoidChannelPromise.
-                closeFuture.addListener(new ChannelFutureListener() {
+                if (!(promise instanceof VoidChannelPromise)) {
-                    @Override
+                    // This means close() was called before so we just register a listener and return
-                    public void operationComplete(ChannelFuture future) throws Exception {
+                    closeFuture.addListener(new ChannelFutureListener() {
-                        promise.setSuccess();
+                        @Override
-                    }
+                        public void operationComplete(ChannelFuture future) throws Exception {
-                });
+                            promise.setSuccess();
+                        }
+                    });
+                }
                 return;
             }
 
@@ -570,19 +563,42 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
                 closeExecutor.execute(new OneTimeTask() {
                     @Override
                     public void run() {
-                        doClose0(promise);
+                        try {
-                        // Call invokeLater so closeAndDeregister is executed in the EventLoop again!
+                            // Execute the close.
-                        invokeLater(new OneTimeTask() {
+                            doClose0(promise);
-                            @Override
+                        } finally {
-                            public void run() {
+                            // Call invokeLater so closeAndDeregister is executed in the EventLoop again!
-                                closeAndDeregister(buffer, wasActive);
+                            invokeLater(new OneTimeTask() {
-                            }
+                                @Override
-                        });
+                                public void run() {
+                                    // Fail all the queued messages
+                                    buffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
+                                    buffer.close(CLOSED_CHANNEL_EXCEPTION);
+                                    fireChannelInactiveAndDeregister(wasActive);
+                                }
+                            });
+                        }
                     }
                 });
             } else {
-                doClose0(promise);
+                try {
-                closeAndDeregister(buffer, wasActive);
+                    // Close the channel and fail the queued messages in all cases.
+                    doClose0(promise);
+                } finally {
+                    // Fail all the queued messages.
+                    buffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
+                    buffer.close(CLOSED_CHANNEL_EXCEPTION);
+                }
+                if (inFlush0) {
+                    invokeLater(new OneTimeTask() {
+                        @Override
+                        public void run() {
+                            fireChannelInactiveAndDeregister(wasActive);
+                        }
+                    });
+                } else {
+                    fireChannelInactiveAndDeregister(wasActive);
+                }
             }
         }
 
@@ -597,23 +613,17 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
             }
         }
 
-        private void closeAndDeregister(ChannelOutboundBuffer outboundBuffer, final boolean wasActive) {
+        private void fireChannelInactiveAndDeregister(final boolean wasActive) {
-            // Fail all the queued messages
+            if (wasActive && !isActive()) {
-            try {
+                invokeLater(new OneTimeTask() {
-                outboundBuffer.failFlushed(CLOSED_CHANNEL_EXCEPTION);
+                    @Override
-                outboundBuffer.close(CLOSED_CHANNEL_EXCEPTION);
+                    public void run() {
-            } finally {
+                        pipeline.fireChannelInactive();
-                if (wasActive && !isActive()) {
+                    }
-                    invokeLater(new OneTimeTask() {
+                });
-                        @Override
-                        public void run() {
-                            pipeline.fireChannelInactive();
-                        }
-                    });
-                }
-
-                deregister(voidPromise());
             }
+
+            deregister(voidPromise());
         }
 
         @Override