renamebrecon/javaReturn

10 files changed, 457 insertions, 42 deletions

diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function.java
deleted file mode 100644
index cb02865d57..0000000000
--- a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function.java
+++ /dev/null
@@ -1,13 +0,0 @@
-// Licensed to the .NET Foundation under one or more agreements.
-// The .NET Foundation licenses this file to you under the MIT license.
-
-package com.microsoft.signalr;
-
-/**
- * A callback that takes no parameters.
- */
-public interface Function<TResult> {
-    // We can't use the @FunctionalInterface annotation because it's only
-    // available on Android API Level 24 and above.
-    TResult invoke();
-}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function1.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function2Single.java
index e00e87bdd0..75d2ba4d2d 100644
--- a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function1.java
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function2Single.java
@@ -3,13 +3,16 @@
 
 package com.microsoft.signalr;
 
+import io.reactivex.rxjava3.core.Single;
+
 /**
  * A callback that takes one parameter.
  *
  * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
  */
-public interface Function1<T1, TResult> {
+public interface Function2Single<T1, T2, TResult> {
     // We can't use the @FunctionalInterface annotation because it's only
     // available on Android API Level 24 and above.
-    TResult invoke(T1 param1);
+    Single<TResult> invoke(T1 param1, T2 param2);
 }
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function3Single.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function3Single.java
new file mode 100644
index 0000000000..027b0380db
--- /dev/null
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function3Single.java
@@ -0,0 +1,19 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+package com.microsoft.signalr;
+
+import io.reactivex.rxjava3.core.Single;
+
+/**
+ * A callback that takes one parameter.
+ *
+ * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
+ * @param <T3> The type of the third parameter to the callback.
+ */
+public interface Function3Single<T1, T2, T3, TResult> {
+    // We can't use the @FunctionalInterface annotation because it's only
+    // available on Android API Level 24 and above.
+    Single<TResult> invoke(T1 param1, T2 param2, T3 param3);
+}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function4Single.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function4Single.java
new file mode 100644
index 0000000000..af009deca1
--- /dev/null
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function4Single.java
@@ -0,0 +1,20 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+package com.microsoft.signalr;
+
+import io.reactivex.rxjava3.core.Single;
+
+/**
+ * A callback that takes one parameter.
+ *
+ * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
+ * @param <T3> The type of the third parameter to the callback.
+ * @param <T4> The type of the fourth parameter to the callback.
+ */
+public interface Function4Single<T1, T2, T3, T4, TResult> {
+    // We can't use the @FunctionalInterface annotation because it's only
+    // available on Android API Level 24 and above.
+    Single<TResult> invoke(T1 param1, T2 param2, T3 param3, T4 param4);
+}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function5Single.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function5Single.java
new file mode 100644
index 0000000000..4776b8b062
--- /dev/null
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function5Single.java
@@ -0,0 +1,21 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+package com.microsoft.signalr;
+
+import io.reactivex.rxjava3.core.Single;
+
+/**
+ * A callback that takes one parameter.
+ *
+ * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
+ * @param <T3> The type of the third parameter to the callback.
+ * @param <T4> The type of the fourth parameter to the callback.
+ * @param <T5> The type of the fifth parameter to the callback.
+ */
+public interface Function5Single<T1, T2, T3, T4, T5, TResult> {
+    // We can't use the @FunctionalInterface annotation because it's only
+    // available on Android API Level 24 and above.
+    Single<TResult> invoke(T1 param1, T2 param2, T3 param3, T4 param4, T5 param5);
+}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function6Single.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function6Single.java
new file mode 100644
index 0000000000..342cda35b0
--- /dev/null
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function6Single.java
@@ -0,0 +1,22 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+package com.microsoft.signalr;
+
+import io.reactivex.rxjava3.core.Single;
+
+/**
+ * A callback that takes one parameter.
+ *
+ * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
+ * @param <T3> The type of the third parameter to the callback.
+ * @param <T4> The type of the fourth parameter to the callback.
+ * @param <T5> The type of the fifth parameter to the callback.
+ * @param <T6> The type of the sixth parameter to the callback.
+ */
+public interface Function6Single<T1, T2, T3, T4, T5, T6, TResult> {
+    // We can't use the @FunctionalInterface annotation because it's only
+    // available on Android API Level 24 and above.
+    Single<TResult> invoke(T1 param1, T2 param2, T3 param3, T4 param4, T5 param5, T6 param6);
+}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function7Single.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function7Single.java
new file mode 100644
index 0000000000..da90e47adc
--- /dev/null
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function7Single.java
@@ -0,0 +1,23 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+package com.microsoft.signalr;
+
+import io.reactivex.rxjava3.core.Single;
+
+/**
+ * A callback that takes one parameter.
+ *
+ * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
+ * @param <T3> The type of the third parameter to the callback.
+ * @param <T4> The type of the fourth parameter to the callback.
+ * @param <T5> The type of the fifth parameter to the callback.
+ * @param <T6> The type of the sixth parameter to the callback.
+ * @param <T7> The type of the seventh parameter to the callback.
+ */
+public interface Function7Single<T1, T2, T3, T4, T5, T6, T7, TResult> {
+    // We can't use the @FunctionalInterface annotation because it's only
+    // available on Android API Level 24 and above.
+    Single<TResult> invoke(T1 param1, T2 param2, T3 param3, T4 param4, T5 param5, T6 param6, T7 param7);
+}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function8Single.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function8Single.java
new file mode 100644
index 0000000000..b8932fef03
--- /dev/null
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/Function8Single.java
@@ -0,0 +1,24 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+package com.microsoft.signalr;
+
+import io.reactivex.rxjava3.core.Single;
+
+/**
+ * A callback that takes one parameter.
+ *
+ * @param <T1> The type of the first parameter to the callback.
+ * @param <T2> The type of the second parameter to the callback.
+ * @param <T3> The type of the third parameter to the callback.
+ * @param <T4> The type of the fourth parameter to the callback.
+ * @param <T5> The type of the fifth parameter to the callback.
+ * @param <T6> The type of the sixth parameter to the callback.
+ * @param <T7> The type of the seventh parameter to the callback.
+ * @param <T8> The type of the eighth parameter to the callback.
+ */
+public interface Function8Single<T1, T2, T3, T4, T5, T6, T7, T8, TResult> {
+    // We can't use the @FunctionalInterface annotation because it's only
+    // available on Android API Level 24 and above.
+    Single<TResult> invoke(T1 param1, T2 param2, T3 param3, T4 param4, T5 param5, T6 param6, T7 param7, T8 param8);
+}
\ No newline at end of file
diff --git a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java
index 185be1e327..b10afb4fd2 100644
--- a/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java
+++ b/src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java
@@ -56,7 +56,6 @@ public class HubConnection implements AutoCloseable {
     // Private property, modified for testing
     private long tickRate = 1000;
 
-
     // Holds all mutable state other than user-defined handlers and settable properties.
     private final ReconnectingConnectionState state;
 
@@ -1297,24 +1296,67 @@ public class HubConnection implements AutoCloseable {
         return registerHandler(target, action, param1, param2, param3, param4, param5, param6, param7, param8);
     }
 
-    public <TResult> Subscription on(String target, Function<TResult> callback) {
-        FunctionBase action = args -> Single.just(callback.invoke());
+    public <TResult> Subscription onWithResult(String target, FunctionSingle<TResult> callback) {
+        FunctionBase action = args -> callback.invoke().cast(Object.class);
         return registerHandler(target, action);
     }
 
-    public <TResult> Subscription on(String target, FunctionSingle<TResult> callback) {
-        FunctionBase action = args -> callback.invoke().cast(Object.class);
-        return registerHandler(target, action);
+    public <T1, TResult> Subscription onWithResult(String target, Function1Single<T1, TResult> callback, Class<T1> param1) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0])).cast(Object.class);
+        return registerHandler(target, action, param1);
     }
 
-    public <T1, TResult> Subscription on(String target, Function1<T1, TResult> callback, Class<T1> param1) {
-        FunctionBase action = params -> Single.just(callback.invoke(Utils.<T1>cast(param1, params[0])));
-        return registerHandler(target, action);
+    public <T1, T2, TResult> Subscription onWithResult(String target, Function2Single<T1, T2, TResult> callback,
+                                             Class<T1> param1, Class<T2> param2) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]),
+            Utils.<T2>cast(param2, params[1])).cast(Object.class);
+        return registerHandler(target, action, param1, param2);
     }
 
-    public <T1, TResult> Subscription on(String target, Function1Single<T1, TResult> callback, Class<T1> param1) {
-        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0])).cast(Object.class);
-        return registerHandler(target, action);
+    public <T1, T2, T3, TResult> Subscription onWithResult(String target, Function3Single<T1, T2, T3, TResult> callback,
+                                                 Class<T1> param1, Class<T2> param2, Class<T3> param3) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]),
+            Utils.<T2>cast(param2, params[1]), Utils.<T3>cast(param3, params[2])).cast(Object.class);
+        return registerHandler(target, action, param1, param2, param3);
+    }
+
+    public <T1, T2, T3, T4, TResult> Subscription onWithResult(String target, Function4Single<T1, T2, T3, T4, TResult> callback,
+                                                     Class<T1> param1, Class<T2> param2, Class<T3> param3, Class<T4> param4) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]), Utils.<T2>cast(param2, params[1]),
+            Utils.<T3>cast(param3, params[2]), Utils.<T4>cast(param4, params[3])).cast(Object.class);
+        return registerHandler(target, action, param1, param2, param3, param4);
+    }
+
+    public <T1, T2, T3, T4, T5, TResult> Subscription onWithResult(String target, Function5Single<T1, T2, T3, T4, T5, TResult> callback,
+                                                         Class<T1> param1, Class<T2> param2, Class<T3> param3, Class<T4> param4, Class<T5> param5) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]), Utils.<T2>cast(param2, params[1]),
+            Utils.<T3>cast(param3, params[2]), Utils.<T4>cast(param4, params[3]), Utils.<T5>cast(param5, params[4])).cast(Object.class);
+        return registerHandler(target, action, param1, param2, param3, param4, param5);
+    }
+
+    public <T1, T2, T3, T4, T5, T6, TResult> Subscription onWithResult(String target, Function6Single<T1, T2, T3, T4, T5, T6, TResult> callback,
+                                                             Class<T1> param1, Class<T2> param2, Class<T3> param3,
+                                                             Class<T4> param4, Class<T5> param5, Class<T6> param6) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]), Utils.<T2>cast(param2, params[1]), Utils.<T3>cast(param3, params[2]),
+            Utils.<T4>cast(param4, params[3]), Utils.<T5>cast(param5, params[4]), Utils.<T6>cast(param6, params[5])).cast(Object.class);
+        return registerHandler(target, action, param1, param2, param3, param4, param5, param6);
+    }
+
+    public <T1, T2, T3, T4, T5, T6, T7, TResult> Subscription onWithResult(String target, Function7Single<T1, T2, T3, T4, T5, T6, T7, TResult> callback,
+                                                                 Class<T1> param1, Class<T2> param2, Class<T3> param3, Class<T4> param4,
+                                                                 Class<T5> param5, Class<T6> param6, Class<T7> param7) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]), Utils.<T2>cast(param2, params[1]), Utils.<T3>cast(param3, params[2]),
+            Utils.<T4>cast(param4, params[3]), Utils.<T5>cast(param5, params[4]), Utils.<T6>cast(param6, params[5]), Utils.<T7>cast(param7, params[6])).cast(Object.class);
+        return registerHandler(target, action, param1, param2, param3, param4, param5, param6, param7);
+    }
+
+    public <T1, T2, T3, T4, T5, T6, T7, T8, TResult> Subscription onWithResult(String target, Function8Single<T1, T2, T3, T4, T5, T6, T7, T8, TResult> callback,
+                                                                     Class<T1> param1, Class<T2> param2, Class<T3> param3, Class<T4> param4, Class<T5> param5,
+                                                                     Class<T6> param6, Class<T7> param7, Class<T8> param8) {
+        FunctionBase action = params -> callback.invoke(Utils.<T1>cast(param1, params[0]), Utils.<T2>cast(param2, params[1]),
+            Utils.<T3>cast(param3, params[2]), Utils.<T4>cast(param4, params[3]), Utils.<T5>cast(param5, params[4]),
+            Utils.<T6>cast(param6, params[5]), Utils.<T7>cast(param7, params[6]), Utils.<T8>cast(param8, params[7])).cast(Object.class);
+        return registerHandler(target, action, param1, param2, param3, param4, param5, param6, param7, param8);
     }
 
     private Subscription registerHandler(String target, Object action, Type... types) {
@@ -1493,7 +1535,7 @@ public class HubConnection implements AutoCloseable {
         }
 
         private void handleInvocations() {
-            messages.observeOn(Schedulers.computation()).subscribe(invocationMessage -> {
+            messages.observeOn(Schedulers.io()).subscribe(invocationMessage -> {
                 List<InvocationHandler> handlers = this.connection.handlers.get(invocationMessage.getTarget());
                 boolean expectsResult = invocationMessage.getInvocationId() != null;
                 if (handlers == null) {
@@ -1543,6 +1585,7 @@ public class HubConnection implements AutoCloseable {
                     logger.warn("Result given for '{}' method but server is not expecting a result.", invocationMessage.getTarget());
                 }
             }, (e) -> {
+                stop(e.getMessage());
             }, () -> {
             });
         }
diff --git a/src/SignalR/clients/java/signalr/test/src/main/java/com/microsoft/signalr/HubConnection.ReturnResultTest.java b/src/SignalR/clients/java/signalr/test/src/main/java/com/microsoft/signalr/HubConnection.ReturnResultTest.java
index b64fb5b7af..1340eae726 100644
--- a/src/SignalR/clients/java/signalr/test/src/main/java/com/microsoft/signalr/HubConnection.ReturnResultTest.java
+++ b/src/SignalR/clients/java/signalr/test/src/main/java/com/microsoft/signalr/HubConnection.ReturnResultTest.java
@@ -8,11 +8,12 @@ import static org.junit.jupiter.api.Assertions.*;
 import java.nio.ByteBuffer;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 import io.reactivex.rxjava3.subjects.CompletableSubject;
 import io.reactivex.rxjava3.subjects.SingleSubject;
 
@@ -26,9 +27,9 @@ class HubConnectionReturnResultTest {
         HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
         AtomicBoolean handlerCalled = new AtomicBoolean();
 
-        hubConnection.on("inc", () -> {
+        hubConnection.onWithResult("inc", () -> {
             handlerCalled.set(true);
-            return 10;
+            return Single.just(10);
         });
 
         hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
@@ -79,16 +80,13 @@ class HubConnectionReturnResultTest {
     public void throwFromReturningOnHandlerWithRequestedResult() {
         MockTransport mockTransport = new MockTransport();
         HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
-        AtomicBoolean handlerCalled = new AtomicBoolean();
-
-        hubConnection.on("inc", () -> {
-            handlerCalled.set(true);
 
+        hubConnection.onWithResult("inc", () -> {
             boolean b = true;
             if (b) {
                 throw new RuntimeException("Custom error.");
             }
-            return "value";
+            return Single.just("value");
         });
 
         hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
@@ -105,13 +103,13 @@ class HubConnectionReturnResultTest {
         MockTransport mockTransport = new MockTransport();
         HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
 
-        hubConnection.on("inc", () -> {
-            return "value";
+        hubConnection.onWithResult("inc", () -> {
+            return Single.just("value");
         });
 
         RuntimeException ex = assertThrows(RuntimeException.class, () -> {
-            hubConnection.on("inc", () -> {
-                return "value2";
+            hubConnection.onWithResult("inc", () -> {
+                return Single.just("value2");
             });
         });
         assertEquals("'inc' already has a value returning handler. Multiple return values are not supported.", ex.getMessage());
@@ -124,8 +122,8 @@ class HubConnectionReturnResultTest {
             MockTransport mockTransport = new MockTransport();
             HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
 
-            hubConnection.on("m", () -> {
-                return 42;
+            hubConnection.onWithResult("m", () -> {
+                return Single.just(42);
             });
 
             hubConnection.on("fin", () -> {
@@ -142,4 +140,259 @@ class HubConnectionReturnResultTest {
             logger.assertLog("Result given for 'm' method but server is not expecting a result.");
         }
     }
+
+    @Test
+    public void returnFromOnHandlerOneParam() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i) -> {
+            handlerCalled.set(i);
+            return Single.just(10);
+        }, String.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[1]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":10}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerTwoParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            return Single.just("bob");
+        }, String.class, Integer.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[1,13]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerThreeParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+        AtomicReference<Integer[]> handler3Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j, k) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            handler3Called.set(k);
+            return Single.just("bob");
+        }, String.class, Integer.class, Integer[].class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[1,13,[1,2,3]]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        assertArrayEquals(new Integer[] { 1, 2, 3 }, handler3Called.get());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerFourParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+        AtomicReference<Integer[]> handler3Called = new AtomicReference<>();
+        AtomicReference<Boolean> handler4Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j, k, l) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            handler3Called.set(k);
+            handler4Called.set(l);
+            return Single.just("bob");
+        }, String.class, Integer.class, Integer[].class, Boolean.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[\"1\",13,[1,2,3],true]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        assertArrayEquals(new Integer[] { 1, 2, 3 }, handler3Called.get());
+        assertEquals(true, handler4Called.get());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerFiveParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+        AtomicReference<Integer[]> handler3Called = new AtomicReference<>();
+        AtomicReference<Boolean> handler4Called = new AtomicReference<>();
+        AtomicReference<String> handler5Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j, k, l, m) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            handler3Called.set(k);
+            handler4Called.set(l);
+            handler5Called.set(m);
+            return Single.just("bob");
+        }, String.class, Integer.class, Integer[].class, Boolean.class, String.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[\"1\",13,[1,2,3],true,\"t\"]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        assertArrayEquals(new Integer[] { 1, 2, 3 }, handler3Called.get());
+        assertEquals(true, handler4Called.get());
+        assertEquals("t", handler5Called.get());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerSixParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+        AtomicReference<Integer[]> handler3Called = new AtomicReference<>();
+        AtomicReference<Boolean> handler4Called = new AtomicReference<>();
+        AtomicReference<String> handler5Called = new AtomicReference<>();
+        AtomicReference<Double> handler6Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j, k, l, m, n) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            handler3Called.set(k);
+            handler4Called.set(l);
+            handler5Called.set(m);
+            handler6Called.set(n);
+            return Single.just("bob");
+        }, String.class, Integer.class, Integer[].class, Boolean.class,
+            String.class, Double.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[\"1\",13,[1,2,3],true,\"t\",1.5]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        assertArrayEquals(new Integer[] { 1, 2, 3 }, handler3Called.get());
+        assertEquals(true, handler4Called.get());
+        assertEquals("t", handler5Called.get());
+        assertEquals(1.5, handler6Called.get().doubleValue());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerSevenParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+        AtomicReference<Integer[]> handler3Called = new AtomicReference<>();
+        AtomicReference<Boolean> handler4Called = new AtomicReference<>();
+        AtomicReference<String> handler5Called = new AtomicReference<>();
+        AtomicReference<Double> handler6Called = new AtomicReference<>();
+        AtomicReference<String> handler7Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j, k, l, m, n, o) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            handler3Called.set(k);
+            handler4Called.set(l);
+            handler5Called.set(m);
+            handler6Called.set(n);
+            handler7Called.set(o);
+            return Single.just("bob");
+        }, String.class, Integer.class, Integer[].class, Boolean.class,
+            String.class, Double.class, String.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[\"1\",13,[1,2,3],true,\"t\",1.5,\"h\"]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        assertArrayEquals(new Integer[] { 1, 2, 3 }, handler3Called.get());
+        assertEquals(true, handler4Called.get());
+        assertEquals("t", handler5Called.get());
+        assertEquals(1.5, handler6Called.get().doubleValue());
+        assertEquals("h", handler7Called.get());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
+
+    @Test
+    public void returnFromOnHandlerEightParams() {
+        MockTransport mockTransport = new MockTransport();
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+        AtomicReference<String> handlerCalled = new AtomicReference<>();
+        AtomicReference<Integer> handler2Called = new AtomicReference<>();
+        AtomicReference<Integer[]> handler3Called = new AtomicReference<>();
+        AtomicReference<Boolean> handler4Called = new AtomicReference<>();
+        AtomicReference<String> handler5Called = new AtomicReference<>();
+        AtomicReference<Double> handler6Called = new AtomicReference<>();
+        AtomicReference<String> handler7Called = new AtomicReference<>();
+        AtomicReference<Integer> handler8Called = new AtomicReference<>();
+
+        hubConnection.onWithResult("inc", (i, j, k, l, m, n, o, p) -> {
+            handlerCalled.set(i);
+            handler2Called.set(j);
+            handler3Called.set(k);
+            handler4Called.set(l);
+            handler5Called.set(m);
+            handler6Called.set(n);
+            handler7Called.set(o);
+            handler8Called.set(p);
+            return Single.just("bob");
+        }, String.class, Integer.class, Integer[].class, Boolean.class,
+            String.class, Double.class, String.class, Integer.class);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+        SingleSubject<ByteBuffer> sendTask = mockTransport.getNextSentMessage();
+        mockTransport.receiveMessage("{\"type\":1,\"invocationId\":\"1\",\"target\":\"inc\",\"arguments\":[\"1\",13,[1,2,3],true,\"t\",1.5,\"h\",33]}" + RECORD_SEPARATOR);
+
+        ByteBuffer message = sendTask.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertEquals("1", handlerCalled.get());
+        assertEquals(13, handler2Called.get().intValue());
+        assertArrayEquals(new Integer[] { 1, 2, 3 }, handler3Called.get());
+        assertEquals(true, handler4Called.get());
+        assertEquals("t", handler5Called.get());
+        assertEquals(1.5, handler6Called.get().doubleValue());
+        assertEquals("h", handler7Called.get());
+        assertEquals(33, handler8Called.get().intValue());
+        String expected = "{\"type\":3,\"invocationId\":\"1\",\"result\":\"bob\"}" + RECORD_SEPARATOR;
+        assertEquals(expected, TestUtils.byteBufferToString(message));
+    }
 }