Refactor RequestHandler.handleChannel to only take a Publisher<Payloa…

…d> as argument. The `initialPayload` was already part of the `payloads` publisher, this was very confusing. + Fix the tests impacted by the new API. + Code style: Ensure lines length < 100 characters
rsocket · Oct 26, 2015 · adb7ce5 · adb7ce5 · benjchristensen · Oct 27, 2015
1 parent 468dadf
commit adb7ce5
Show file tree

Hide file tree

Showing 13 changed files with 1,852 additions and 1,511 deletions.
diff --git a/src/main/java/io/reactivesocket/DuplexConnection.java b/src/main/java/io/reactivesocket/DuplexConnection.java
@@ -31,5 +31,11 @@ public interface DuplexConnection extends Closeable {
 	Observable<Frame> getInput();
 
 	void addOutput(Publisher<Frame> o, Completable callback);
-
+
+	default void addOutput(Frame frame, Completable callback) {
+        addOutput(s -> {
+            s.onNext(frame);
+            s.onComplete();
+        }, callback);
+	}
 }