Delegate dispose/isDisposed to underlying channel in NettyDuplexConnection

Author: rdegnan

gradle.properties

@@ -14,6 +14,6 @@
 # limitations under the License.
 #
 
-mavenversion=0.9-SNAPSHOT
+mavenversion=0.10-SNAPSHOT
 release.scope=patch
-release.version=0.9-SNAPSHOT
+release.version=0.10-SNAPSHOT

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

@@ -20,7 +20,6 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.ipc.netty.NettyContext;
 import reactor.ipc.netty.NettyInbound;
 import reactor.ipc.netty.NettyOutbound;
@@ -29,22 +28,11 @@ public class NettyDuplexConnection implements DuplexConnection {
   private final NettyInbound in;
   private final NettyOutbound out;
   private final NettyContext context;
-  private final MonoProcessor<Void> onClose;
 
   public NettyDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) {
     this.in = in;
     this.out = out;
     this.context = context;
-    this.onClose = MonoProcessor.create();
-
-    context.onClose(onClose::onComplete);
-    this.onClose
-        .doFinally(
-            s -> {
-              this.context.dispose();
-              this.context.channel().close();
-            })
-        .subscribe();
   }
 
   @Override
@@ -64,16 +52,16 @@ public Flux<Frame> receive() {
 
   @Override
   public void dispose() {
-    onClose.onComplete();
+    context.dispose();
   }
 
   @Override
   public boolean isDisposed() {
-    return onClose.isDisposed();
+    return context.isDisposed();
   }
 
   @Override
   public Mono<Void> onClose() {
-    return onClose;
+    return context.onClose();
   }
 }

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

@@ -27,7 +27,6 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.ipc.netty.NettyContext;
 import reactor.ipc.netty.NettyInbound;
 import reactor.ipc.netty.NettyOutbound;
@@ -43,22 +42,11 @@ public class WebsocketDuplexConnection implements DuplexConnection {
   private final NettyInbound in;
   private final NettyOutbound out;
   private final NettyContext context;
-  private final MonoProcessor<Void> onClose;
 
   public WebsocketDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) {
     this.in = in;
     this.out = out;
     this.context = context;
-    this.onClose = MonoProcessor.create();
-
-    context.onClose(onClose::onComplete);
-    this.onClose
-        .doFinally(
-            s -> {
-              this.context.dispose();
-              this.context.channel().close();
-            })
-        .subscribe();
   }
 
   @Override
@@ -87,16 +75,16 @@ public Flux<Frame> receive() {
 
   @Override
   public void dispose() {
-    onClose.onComplete();
+    context.dispose();
   }
 
   @Override
   public boolean isDisposed() {
-    return onClose.isDisposed();
+    return context.isDisposed();
   }
 
   @Override
   public Mono<Void> onClose() {
-    return onClose;
+    return context.onClose();
   }
 }