Make sure ChannelFutureListeners are invoked from an event loop thread

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

@@ -15,11 +15,11 @@
  */
 package io.netty.channel;
 
-import java.util.concurrent.TimeUnit;
-
 import io.netty.logging.InternalLogger;
 import io.netty.logging.InternalLoggerFactory;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * A skeletal {@link ChannelFuture} implementation which represents a
  * {@link ChannelFuture} which has been completed already.
@@ -44,7 +44,20 @@ public abstract class CompleteChannelFuture implements ChannelFuture {
     }
 
     @Override
-    public void addListener(ChannelFutureListener listener) {
+    public void addListener(final ChannelFutureListener listener) {
+        if (channel().eventLoop().inEventLoop()) {
+            notifyListener(listener);
+        } else {
+            channel().eventLoop().execute(new Runnable() {
+                @Override
+                public void run() {
+                    notifyListener(listener);
+                }
+            });
+        }
+    }
+
+    private void notifyListener(ChannelFutureListener listener) {
         try {
             listener.operationComplete(this);
         } catch (Throwable t) {

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

@@ -16,16 +16,16 @@
 package io.netty.channel;
 
 import static java.util.concurrent.TimeUnit.*;
+import io.netty.logging.InternalLogger;
+import io.netty.logging.InternalLoggerFactory;
+import io.netty.util.internal.DeadLockProofWorker;
 
+import java.nio.channels.Channels;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
-import io.netty.logging.InternalLogger;
-import io.netty.logging.InternalLoggerFactory;
-import io.netty.util.internal.DeadLockProofWorker;
-
 /**
  * The default {@link ChannelFuture} implementation.  It is recommended to
  * use {@link Channels#future(Channel)} and {@link Channels#future(Channel, boolean)}
@@ -119,7 +119,7 @@ public class DefaultChannelFuture implements ChannelFuture {
     }
 
     @Override
-    public void addListener(ChannelFutureListener listener) {
+    public void addListener(final ChannelFutureListener listener) {
         if (listener == null) {
             throw new NullPointerException("listener");
         }
@@ -148,7 +148,17 @@ public class DefaultChannelFuture implements ChannelFuture {
         }
 
         if (notifyNow) {
+            if (channel().eventLoop().inEventLoop()) {
                 notifyListener(listener);
+            } else {
+                channel().eventLoop().execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        notifyListener(listener);
+
+                    }
+                });
+            }
         }
     }
 
@@ -398,7 +408,12 @@ public class DefaultChannelFuture implements ChannelFuture {
         //    Hence any listener list modification happens-before this method.
         // 2) This method is called only when 'done' is true.  Once 'done'
         //    becomes true, the listener list is never modified - see add/removeListener()
-        if (firstListener != null) {
+
+        if (firstListener == null) {
+            return;
+        }
+
+        if (channel().eventLoop().inEventLoop()) {
             notifyListener(firstListener);
             firstListener = null;
 
@@ -408,6 +423,22 @@ public class DefaultChannelFuture implements ChannelFuture {
                 }
                 otherListeners = null;
             }
+        } else {
+            final ChannelFutureListener firstListener = this.firstListener;
+            final List<ChannelFutureListener> otherListeners = this.otherListeners;
+            this.firstListener = null;
+            this.otherListeners = null;
+            channel().eventLoop().execute(new Runnable() {
+                @Override
+                public void run() {
+                    notifyListener(firstListener);
+                    if (otherListeners != null) {
+                        for (ChannelFutureListener l: otherListeners) {
+                            notifyListener(l);
+                        }
+                    }
+                }
+            });
         }
     }