Revert refactor RequestHandler.handleChannel() to take two arguments

Author: stevegury

build.gradle

@@ -25,7 +25,6 @@ dependencies {
 	testCompile 'org.mockito:mockito-core:1.8.5'
 }
 
-
 // support for snapshot/final releases via versioned branch names like 1.x
 nebulaRelease {
     addReleaseBranchPattern(/\d+\.\d+\.\d+/)
@@ -38,4 +37,4 @@ if (project.hasProperty('release.useLastTag')) {
 
 test {
     testLogging.showStandardStreams = true
-}
+}

src/main/java/io/reactivesocket/RequestHandler.java

@@ -18,7 +18,6 @@
 import io.reactivesocket.internal.PublisherUtils;
 import org.reactivestreams.Publisher;
 
-import java.util.function.BiFunction;
 import java.util.function.Function;
 
 public abstract class RequestHandler {
@@ -50,9 +49,10 @@ public abstract class RequestHandler {
 	public abstract Publisher<Void> handleFireAndForget(final Payload payload);
 
 	/**
-	 * @return
+	 * @note The initialPayload will also be part of the inputs publisher.
+	 * It is there to simplify routing logic.
 	 */
-	public abstract Publisher<Payload> handleChannel(final Publisher<Payload> inputs);
+	public abstract Publisher<Payload> handleChannel(Payload initialPayload, final Publisher<Payload> inputs);
 
 	public abstract Publisher<Void> handleMetadataPush(final Payload payload);
 
@@ -125,7 +125,7 @@ public Publisher<Void> handleFireAndForget(Payload payload)
 					return handleFireAndForget.apply(payload);
 				}
 
-				public Publisher<Payload> handleChannel(Publisher<Payload> inputs)
+				public Publisher<Payload> handleChannel(Payload initialPayload, Publisher<Payload> inputs)
 				{
 					return handleRequestChannel.apply(inputs);
 				}

src/main/java/io/reactivesocket/internal/Responder.java

@@ -742,7 +742,7 @@ public void request(long n) {
  									channels.put(streamId, channelRequests);
 								}
 
-                                Publisher<Payload> responses = requestHandler.handleChannel(channelRequests);
+                                Publisher<Payload> responses = requestHandler.handleChannel(requestFrame, channelRequests);
                                 responses.subscribe(new Subscriber<Payload>() {
 									@Override
 									public void onSubscribe(Subscription s) {

src/perf/java/io/reactivesocket/ReactiveSocketPerf.java

@@ -122,7 +122,7 @@ public Publisher<Void> handleFireAndForget(Payload payload) {
 			}
 
 			@Override
-			public Publisher<Payload> handleChannel(Publisher<Payload> inputs) {
+			public Publisher<Payload> handleChannel(Payload initialPayload, Publisher<Payload> inputs) {
 				return null;
 			}
 

src/test/java/io/reactivesocket/LeaseTest.java

@@ -114,7 +114,7 @@ public Publisher<Void> handleFireAndForget(Payload payload) {
                  */
                 @Override
                 public Publisher<Payload> handleChannel(
-                    Publisher<Payload> inputs
+                    Payload initialPayload, Publisher<Payload> inputs
                 ) {
                     return fromPublisher(inputs).map(p ->
                         utf8EncodedPayload(byteToString(p.getData()) + "_echo", null));

src/test/java/io/reactivesocket/ReactiveSocketTest.java

@@ -125,7 +125,7 @@ public Publisher<Void> handleFireAndForget(Payload payload) {
 			 * Use Payload.metadata for routing
 			 */
 			@Override
-			public Publisher<Payload> handleChannel(Publisher<Payload> inputs) {
+			public Publisher<Payload> handleChannel(Payload initialPayload, Publisher<Payload> inputs) {
 				return new Publisher<Payload>() {
 					@Override
 					public void subscribe(Subscriber<? super Payload> subscriber) {