Release payloads by default in AbstractRSocket

Author: rdegnan

rsocket-core/src/main/java/io/rsocket/AbstractRSocket.java

@@ -31,16 +31,19 @@ public abstract class AbstractRSocket implements RSocket {
 
   @Override
   public Mono<Void> fireAndForget(Payload payload) {
+    payload.release();
     return Mono.error(new UnsupportedOperationException("Fire and forget not implemented."));
   }
 
   @Override
   public Mono<Payload> requestResponse(Payload payload) {
+    payload.release();
     return Mono.error(new UnsupportedOperationException("Request-Response not implemented."));
   }
 
   @Override
   public Flux<Payload> requestStream(Payload payload) {
+    payload.release();
     return Flux.error(new UnsupportedOperationException("Request-Stream not implemented."));
   }
 
@@ -51,6 +54,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
 
   @Override
   public Mono<Void> metadataPush(Payload payload) {
+    payload.release();
     return Mono.error(new UnsupportedOperationException("Metadata-Push not implemented."));
   }
 

rsocket-core/src/main/java/io/rsocket/RSocketFactory.java

@@ -248,7 +248,7 @@ public Mono<RSocket> start() {
   }
 
   public static class ServerRSocketFactory {
-    private Supplier<SocketAcceptor> acceptor;
+    private SocketAcceptor acceptor;
     private Function<Frame, ? extends Payload> frameDecoder = DefaultPayload::create;
     private Consumer<Throwable> errorConsumer = Throwable::printStackTrace;
     private int mtu = 0;
@@ -272,11 +272,6 @@ public ServerRSocketFactory addServerPlugin(RSocketInterceptor interceptor) {
     }
 
     public ServerTransportAcceptor acceptor(SocketAcceptor acceptor) {
-      this.acceptor = () -> acceptor;
-      return ServerStart::new;
-    }
-
-    public ServerTransportAcceptor acceptor(Supplier<SocketAcceptor> acceptor) {
       this.acceptor = acceptor;
       return ServerStart::new;
     }
@@ -350,7 +345,6 @@ private Mono<Void> processSetupFrame(
         RSocket wrappedRSocketClient = plugins.applyClient(rSocketClient);
 
         return acceptor
-            .get()
             .accept(setupPayload, wrappedRSocketClient)
             .doOnNext(
                 unwrappedServerSocket -> {