changes onClose subscription style

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java

@@ -11,7 +11,7 @@ public abstract class BaseDuplexConnection implements DuplexConnection {
   protected UnboundedProcessor sender = new UnboundedProcessor();
 
   public BaseDuplexConnection() {
-    onClose.doFinally(s -> doOnClose()).subscribe();
+    onClose.subscribe(null, t -> doOnClose(), this::doOnClose);
   }
 
   @Override

rsocket-core/src/main/java/io/rsocket/keepalive/KeepAliveHandler.java

@@ -27,7 +27,9 @@ public KeepAliveFramesAcceptor start(
         KeepAliveSupport keepAliveSupport,
         Consumer<ByteBuf> onSendKeepAliveFrame,
         Consumer<KeepAlive> onTimeout) {
-      duplexConnection.onClose().doFinally(s -> keepAliveSupport.stop()).subscribe();
+      duplexConnection
+          .onClose()
+          .subscribe(null, __ -> keepAliveSupport.stop(), keepAliveSupport::stop);
       return keepAliveSupport
           .onSendKeepAliveFrame(onSendKeepAliveFrame)
           .onTimeout(onTimeout)