provides reworked AeronTransport implementation

build.gradle

@@ -45,6 +45,7 @@ subprojects {
     ext['assertj.version'] = '3.19.0'
     ext['netflix.limits.version'] = '0.3.6'
     ext['bouncycastle-bcpkix.version'] = '1.68'
+    ext['aeron.version'] = '1.31.1'
 
     group = "io.rsocket"
 
@@ -87,6 +88,10 @@ subprojects {
                 entry 'jmh-core'
                 entry 'jmh-generator-annprocess'
             }
+            dependencySet(group: 'io.aeron', version: ext['aeron.version']) {
+                entry 'aeron-client'
+                entry 'aeron-driver'
+            }
         }
         generatedPomCustomization {
             enabled = false

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

@@ -27,7 +27,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)