Release payloads by default in AbstractRSocket

rdegnan · rdegnan · commit c4fdd9a70d0a · 2018-09-17T15:35:10.000-07:00
diff --git a/rsocket-core/src/main/java/io/rsocket/AbstractRSocket.java b/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."));
   }
 
diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketFactory.java b/rsocket-core/src/main/java/io/rsocket/RSocketFactory.java
@@ -253,7 +253,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;
@@ -277,11 +277,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;
     }
@@ -357,7 +352,6 @@ private Mono<Void> processSetupFrame(
         RSocket wrappedRSocketClient = plugins.applyClient(rSocketClient);
 
         return acceptor
-            .get()
             .accept(setupPayload, wrappedRSocketClient)
             .doOnNext(
                 unwrappedServerSocket -> {

Original file line number	Diff line number	Diff line change
`@@ -31,16 +31,19 @@ public abstract class AbstractRSocket implements RSocket {`
`31`	`31`
`32`	`32`	`@Override`
`33`	`33`	`public Mono<Void> fireAndForget(Payload payload) {`
	`34`	`+ payload.release();`
`34`	`35`	`return Mono.error(new UnsupportedOperationException("Fire and forget not implemented."));`
`35`	`36`	`}`
`36`	`37`
`37`	`38`	`@Override`
`38`	`39`	`public Mono<Payload> requestResponse(Payload payload) {`
	`40`	`+ payload.release();`
`39`	`41`	`return Mono.error(new UnsupportedOperationException("Request-Response not implemented."));`
`40`	`42`	`}`
`41`	`43`
`42`	`44`	`@Override`
`43`	`45`	`public Flux<Payload> requestStream(Payload payload) {`
	`46`	`+ payload.release();`
`44`	`47`	`return Flux.error(new UnsupportedOperationException("Request-Stream not implemented."));`
`45`	`48`	`}`
`46`	`49`
`@@ -51,6 +54,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {`
`51`	`54`
`52`	`55`	`@Override`
`53`	`56`	`public Mono<Void> metadataPush(Payload payload) {`
	`57`	`+ payload.release();`
`54`	`58`	`return Mono.error(new UnsupportedOperationException("Metadata-Push not implemented."));`
`55`	`59`	`}`
`56`	`60`