updates

Author: robertroeser

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

@@ -25,7 +25,7 @@
 import io.rsocket.framing.FrameType;
 
 /**
- * Exposed to server for determination of RequestHandler based on mime types and SETUP metadata/data
+ * Exposed to server for determination of ResponderRSocket based on mime types and SETUP metadata/data
  */
 public abstract class ConnectionSetupPayload extends AbstractReferenceCounted implements Payload {
 

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

@@ -42,12 +42,11 @@
 import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M;
 
 /** Server side RSocket. Receives {@link Frame}s from a {@link RSocketClient} */
-class RSocketServer implements RequestHandler {
+class RSocketServer implements ResponderRSocket {
 
   private final DuplexConnection connection;
   private final RSocket requestHandler;
-  private final RequestHandler optimizedRequestHandler;
-  private final boolean hasOptimizedRequestHandler;
+  private final ResponderRSocket responderRSocket;
   private final Function<Frame, ? extends Payload> frameDecoder;
   private final Consumer<Throwable> errorConsumer;
 
@@ -74,16 +73,10 @@ class RSocketServer implements RequestHandler {
       Consumer<Throwable> errorConsumer,
       long tickPeriod,
       long ackTimeout) {
-    
-    if (requestHandler instanceof RequestHandler) {
-      this.optimizedRequestHandler = (RequestHandler) requestHandler;
-      this.hasOptimizedRequestHandler = true;
-      this.requestHandler = null;
-    } else {
-      this.hasOptimizedRequestHandler = false;
-      this.requestHandler = requestHandler;
-      this.optimizedRequestHandler = null;
-    }
+
+    this.requestHandler = requestHandler;
+    this.responderRSocket =
+        (requestHandler instanceof ResponderRSocket) ? (ResponderRSocket) requestHandler : null;
 
     this.connection = connection;
     this.frameDecoder = frameDecoder;
@@ -221,7 +214,7 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
   @Override
   public Flux<Payload> requestChannel(Payload payload, Publisher<Payload> payloads) {
     try {
-      return optimizedRequestHandler.requestChannel(payloads);
+      return responderRSocket.requestChannel(payloads);
     } catch (Throwable t) {
       return Flux.error(t);
     }
@@ -415,7 +408,7 @@ private void handleChannel(int streamId, Payload payload, int initialRequestN) {
     // and any later payload can be processed
     frames.onNext(payload);
 
-    if (hasOptimizedRequestHandler) {
+    if (responderRSocket != null) {
       handleStream(streamId, requestChannel(payload, payloads), initialRequestN);
     } else {
       handleStream(streamId, requestChannel(payloads), initialRequestN);

rsocket-core/src/main/java/io/rsocket/RequestHandler.java renamed to rsocket-core/src/main/java/io/rsocket/ResponderRSocket.java

@@ -7,7 +7,7 @@
  * Extends the {@link RSocket} that allows an implementer to peek at the first request payload of a
  * channel.
  */
-public interface RequestHandler extends RSocket {
+public interface ResponderRSocket extends RSocket {
   /**
    * Implement this method to peak at the first payload of the incoming request stream without
    * having to subscribe to Publish<Payload> payloads