Acquire HubConnectionStateLock before Send/Invoke/Stream (#12078)

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/HubConnection.java

@@ -537,11 +537,15 @@ public class HubConnection {
      * @param args   The arguments to be passed to the method.
      */
     public void send(String method, Object... args) {
+        hubConnectionStateLock.lock();
+        try {
             if (hubConnectionState != HubConnectionState.CONNECTED) {
                 throw new RuntimeException("The 'send' method cannot be called if the connection is not active.");
             }
-
             sendInvocationMessage(method, args);
+        } finally {
+            hubConnectionStateLock.unlock();
+        }
     }
 
     private void sendInvocationMessage(String method, Object[] args) {
@@ -605,6 +609,8 @@ public class HubConnection {
      */
     @SuppressWarnings("unchecked")
     public Completable invoke(String method, Object... args) {
+        hubConnectionStateLock.lock();
+        try {
             if (hubConnectionState != HubConnectionState.CONNECTED) {
                 throw new RuntimeException("The 'invoke' method cannot be called if the connection is not active.");
             }
@@ -625,6 +631,9 @@ public class HubConnection {
             // where the map doesn't have the callbacks yet when the response is returned
             sendInvocationMessage(method, args, id, false);
             return subject;
+        } finally {
+            hubConnectionStateLock.unlock();
+        }
     }
 
     /**
@@ -638,16 +647,18 @@ public class HubConnection {
      */
     @SuppressWarnings("unchecked")
     public <T> Single<T> invoke(Class<T> returnType, String method, Object... args) {
+        hubConnectionStateLock.lock();
+        try {
             if (hubConnectionState != HubConnectionState.CONNECTED) {
                 throw new RuntimeException("The 'invoke' method cannot be called if the connection is not active.");
             }
 
             String id = connectionState.getNextInvocationId();
-
-        SingleSubject<T> subject = SingleSubject.create();
             InvocationRequest irq = new InvocationRequest(returnType, id);
             connectionState.addInvocation(irq);
 
+            SingleSubject<T> subject = SingleSubject.create();
+
             // forward the invocation result or error to the user
             // run continuations on a separate thread
             Subject<Object> pendingCall = irq.getPendingCall();
@@ -664,6 +675,9 @@ public class HubConnection {
             // where the map doesn't have the callbacks yet when the response is returned
             sendInvocationMessage(method, args, id, false);
             return subject;
+        } finally {
+            hubConnectionStateLock.unlock();
+        }
     }
 
     /**
@@ -677,12 +691,20 @@ public class HubConnection {
      */
     @SuppressWarnings("unchecked")
     public <T> Observable<T> stream(Class<T> returnType, String method, Object ... args) {
-        String invocationId = connectionState.getNextInvocationId();
+        String invocationId;
-        AtomicInteger subscriptionCount = new AtomicInteger();
+        InvocationRequest irq;
-        InvocationRequest irq = new InvocationRequest(returnType, invocationId);
+        hubConnectionStateLock.lock();
-        connectionState.addInvocation(irq);
+        try {
-        ReplaySubject<T> subject = ReplaySubject.create();
+            if (hubConnectionState != HubConnectionState.CONNECTED) {
+                throw new RuntimeException("The 'stream' method cannot be called if the connection is not active.");
+            }
 
+            invocationId = connectionState.getNextInvocationId();
+            irq = new InvocationRequest(returnType, invocationId);
+            connectionState.addInvocation(irq);
+
+            AtomicInteger subscriptionCount = new AtomicInteger();
+            ReplaySubject<T> subject = ReplaySubject.create();
             Subject<Object> pendingCall = irq.getPendingCall();
             pendingCall.subscribe(result -> {
                         // Primitive types can't be cast with the Class cast function
@@ -700,10 +722,15 @@ public class HubConnection {
                 if (subscriptionCount.decrementAndGet() == 0) {
                     CancelInvocationMessage cancelInvocationMessage = new CancelInvocationMessage(invocationId);
                     sendHubMessage(cancelInvocationMessage);
+                    if (connectionState != null) {
                         connectionState.tryRemoveInvocation(invocationId);
+                    }
                     subject.onComplete();
                 }
             });
+        } finally {
+            hubConnectionStateLock.unlock();
+        }
     }
 
     private void sendHubMessage(HubMessage message) {

src/SignalR/clients/java/signalr/src/test/java/com/microsoft/signalr/HubConnectionTest.java

@@ -1608,6 +1608,15 @@ class HubConnectionTest {
         assertEquals("The 'invoke' method cannot be called if the connection is not active.", exception.getMessage());
     }
 
+    @Test
+    public void cannotStreamBeforeStart()  {
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com");
+        assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
+
+        Throwable exception = assertThrows(RuntimeException.class, () -> hubConnection.stream(String.class, "inc", "arg1"));
+        assertEquals("The 'stream' method cannot be called if the connection is not active.", exception.getMessage());
+    }
+
     @Test
     public void doesNotErrorWhenReceivingInvokeWithIncorrectArgumentLength()  {
         MockTransport mockTransport = new MockTransport();
@@ -2036,7 +2045,7 @@ class HubConnectionTest {
 
         TestHttpClient client = new TestHttpClient()
                 .on("POST", "http://example.com/negotiate", (req) -> {
-                    if(redirectCount.get() == 0){
+                    if (redirectCount.get() == 0) {
                         redirectCount.incrementAndGet();
                         redirectToken.set(req.getHeaders().get("Authorization"));
                         return Single.just(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\",\"accessToken\":\"firstRedirectToken\"}"));

src/SignalR/clients/java/signalr/src/test/java/com/microsoft/signalr/WebSocketTransportUrlFormatTest.java

@@ -8,7 +8,6 @@ import static org.junit.jupiter.api.Assertions.*;
 import java.util.HashMap;
 import java.util.stream.Stream;
 
-import io.reactivex.Single;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;

src/SignalR/clients/java/signalr/src/test/java/com/microsoft/signalr/sample/Chat.java

@@ -37,6 +37,6 @@ public class Chat {
             hubConnection.send("Send", message);
         }
 
-        hubConnection.stop();
+        hubConnection.stop().blockingAwait();
     }
 }