Flush task should not flush messages that were written since last flu…

…sh attempt.

Motivation:

The flush task is currently using flush() which will have the affect of have the flush traverse the whole ChannelPipeline and also flush messages that were written since we gave up flushing. This is not really correct as we should only continue to flush messages that were flushed at the point in time when the flush task was submitted for execution if the user not explicit call flush() by him/herself.

Modification:

Call *Unsafe.flush0() via the flush task which will only continue flushing messages that were marked as flushed before.

Result:

More correct behaviour when the flush task is used.

transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java

@@ -73,7 +73,9 @@ public abstract class AbstractEpollStreamChannel extends AbstractEpollChannel im
     private final Runnable flushTask = new Runnable() {
         @Override
         public void run() {
-            flush();
+            // Calling flush0 directly to ensure we not try to flush messages that were added via write(...) in the
+            // meantime.
+            ((AbstractEpollUnsafe) unsafe()).flush0();
         }
     };
     private Queue<SpliceInTask> spliceQueue;

transport-native-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueStreamChannel.java

@@ -59,7 +59,9 @@ public abstract class AbstractKQueueStreamChannel extends AbstractKQueueChannel
     private final Runnable flushTask = new Runnable() {
         @Override
         public void run() {
-            flush();
+            // Calling flush0 directly to ensure we not try to flush messages that were added via write(...) in the
+            // meantime.
+            ((AbstractKQueueUnsafe) unsafe()).flush0();
         }
     };
 

transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java

@@ -49,7 +49,9 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
     private final Runnable flushTask = new Runnable() {
         @Override
         public void run() {
-            flush();
+            // Calling flush0 directly to ensure we not try to flush messages that were added via write(...) in the
+            // meantime.
+            ((AbstractNioUnsafe) unsafe()).flush0();
         }
     };