Merge pull request #618 from rsocket/channel-hang-fix

fix issue when long running channels eventually hang

Author: mostroverkhov

rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/SendPublisher.java

@@ -2,7 +2,6 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelPromise;
 import io.netty.channel.EventLoop;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCounted;
@@ -85,26 +84,22 @@ class SendPublisher<V extends ReferenceCounted> extends Flux<ByteBuf> {
     fuse = queue instanceof Fuseable.QueueSubscription;
   }
 
-  private ChannelPromise writeCleanupPromise(V poll) {
-    return channel
-        .newPromise()
-        .addListener(
-            future -> {
-              if (requested != Long.MAX_VALUE) {
-                requested--;
-              }
-              requestedUpstream--;
-              pending--;
+  @SuppressWarnings("unchecked")
+  private void writeCleanup(V poll) {
+    if (requested != Long.MAX_VALUE) {
+      requested--;
+    }
+    requestedUpstream--;
+    pending--;
 
-              InnerSubscriber is = (InnerSubscriber) INNER_SUBSCRIBER.get(SendPublisher.this);
-              if (is != null) {
-                is.tryRequestMoreUpstream();
-                tryComplete(is);
-              }
-              if (poll.refCnt() > 0) {
-                ReferenceCountUtil.safeRelease(poll);
-              }
-            });
+    InnerSubscriber is = (InnerSubscriber) INNER_SUBSCRIBER.get(SendPublisher.this);
+    if (is != null) {
+      is.tryRequestMoreUpstream();
+      tryComplete(is);
+    }
+    if (poll.refCnt() > 0) {
+      ReferenceCountUtil.safeRelease(poll);
+    }
   }
 
   private void tryComplete(InnerSubscriber is) {
@@ -207,7 +202,7 @@ private void flush() {
     }
 
     private void tryDrain() {
-      if (wip == 0 && terminated == 0 && WIP.getAndIncrement(SendPublisher.this) == 0) {
+      if (terminated == 0 && WIP.getAndIncrement(SendPublisher.this) == 0) {
         try {
           if (eventLoop.inEventLoop()) {
             drain();
@@ -235,11 +230,29 @@ private void drain() {
               int readableBytes = sizeOf.size(poll);
               pending++;
               if (channel.isWritable() && readableBytes <= channel.bytesBeforeUnwritable()) {
-                channel.write(poll, writeCleanupPromise(poll));
+                channel
+                    .write(poll)
+                    .addListener(
+                        future -> {
+                          if (future.cause() != null) {
+                            onError(future.cause());
+                          } else {
+                            writeCleanup(poll);
+                          }
+                        });
                 scheduleFlush = true;
               } else {
                 scheduleFlush = false;
-                channel.writeAndFlush(poll, writeCleanupPromise(poll));
+                channel
+                    .writeAndFlush(poll)
+                    .addListener(
+                        future -> {
+                          if (future.cause() != null) {
+                            onError(future.cause());
+                          } else {
+                            writeCleanup(poll);
+                          }
+                        });
               }
 
               tryRequestMoreUpstream();