backport accessTokenProvider fix (#8726)

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

@@ -44,6 +44,7 @@ public class HubConnection {
     private List<OnClosedCallback> onClosedCallbackList;
     private final boolean skipNegotiate;
     private Single<String> accessTokenProvider;
+    private Single<String> redirectAccessTokenProvider;
     private final Map<String, String> headers = new HashMap<>();
     private ConnectionState connectionState = null;
     private final HttpClient httpClient;
@@ -240,11 +241,11 @@ public class HubConnection {
             }
 
             if (negotiateResponse.getAccessToken() != null) {
-                this.accessTokenProvider = Single.just(negotiateResponse.getAccessToken());
+                this.redirectAccessTokenProvider = Single.just(negotiateResponse.getAccessToken());
                 String token = "";
                 // We know the Single is non blocking in this case
                 // It's fine to call blockingGet() on it.
-                token = this.accessTokenProvider.blockingGet();
+                token = this.redirectAccessTokenProvider.blockingGet();
                 this.headers.put("Authorization", "Bearer " + token);
             }
 
@@ -433,6 +434,8 @@ public class HubConnection {
             logger.info("HubConnection stopped.");
             hubConnectionState = HubConnectionState.DISCONNECTED;
             handshakeResponseSubject.onComplete();
+            redirectAccessTokenProvider = null;
+            this.headers.remove("Authorization");
         } finally {
             hubConnectionStateLock.unlock();
         }

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

@@ -1213,4 +1213,86 @@ class HubConnectionTest {
             () -> hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait());
         assertEquals("Unexpected status code returned from negotiate: 500 Internal server error.", exception.getMessage());
     }
+
+    @Test
+    public void accessTokenProviderReferenceIsKeptAfterNegotiateRedirect() {
+        AtomicReference<String> token = new AtomicReference<>();
+        AtomicReference<String> beforeRedirectToken = new AtomicReference<>();
+
+        TestHttpClient client = new TestHttpClient()
+                .on("POST", "http://example.com/negotiate", (req) -> {
+                    beforeRedirectToken.set(req.getHeaders().get("Authorization"));
+                    return Single.just(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\",\"accessToken\":\"newToken\"}"));
+                })
+                .on("POST", "http://testexample.com/negotiate", (req) -> {
+                    token.set(req.getHeaders().get("Authorization"));
+                    return Single.just(new HttpResponse(200, "", "{\"connectionId\":\"bVOiRPG8-6YiJ6d7ZcTOVQ\",\""
+                            + "availableTransports\":[{\"transport\":\"WebSockets\",\"transferFormats\":[\"Text\",\"Binary\"]}]}"));
+                });
+
+        MockTransport transport = new MockTransport(true);
+        HubConnection hubConnection = HubConnectionBuilder
+                .create("http://example.com")
+                .withTransport(transport)
+                .withHttpClient(client)
+                .withAccessTokenProvider(Single.just("User Registered Token"))
+                .build();
+
+        hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
+        assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
+        hubConnection.stop().timeout(1, TimeUnit.SECONDS).blockingAwait();
+        assertEquals("Bearer User Registered Token", beforeRedirectToken.get());
+        assertEquals("Bearer newToken", token.get());
+
+        // Clear the tokens to see if they get reset to the proper values
+        beforeRedirectToken.set("");
+        token.set("");
+
+        // Restart the connection to make sure that the orignal accessTokenProvider that we registered is still registered before the redirect.
+        hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
+        assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
+        hubConnection.stop();
+        assertEquals("Bearer User Registered Token", beforeRedirectToken.get());
+        assertEquals("Bearer newToken", token.get());
+    }
+
+    @Test
+    public void authorizationHeaderFromNegotiateGetsClearedAfterStopping() {
+        AtomicReference<String> token = new AtomicReference<>();
+        AtomicReference<String> beforeRedirectToken = new AtomicReference<>();
+
+        TestHttpClient client = new TestHttpClient()
+                .on("POST", "http://example.com/negotiate", (req) -> {
+                    beforeRedirectToken.set(req.getHeaders().get("Authorization"));
+                    return Single.just(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\",\"accessToken\":\"newToken\"}"));
+                })
+                .on("POST", "http://testexample.com/negotiate", (req) -> {
+                    token.set(req.getHeaders().get("Authorization"));
+                    return Single.just(new HttpResponse(200, "", "{\"connectionId\":\"bVOiRPG8-6YiJ6d7ZcTOVQ\",\""
+                            + "availableTransports\":[{\"transport\":\"WebSockets\",\"transferFormats\":[\"Text\",\"Binary\"]}]}"));
+                });
+
+        MockTransport transport = new MockTransport(true);
+        HubConnection hubConnection = HubConnectionBuilder
+                .create("http://example.com")
+                .withTransport(transport)
+                .withHttpClient(client)
+                .build();
+
+        hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
+        assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
+        hubConnection.stop().timeout(1, TimeUnit.SECONDS).blockingAwait();
+        assertEquals("Bearer newToken", token.get());
+
+        // Clear the tokens to see if they get reset to the proper values
+        beforeRedirectToken.set("");
+        token.set("");
+
+        // Restart the connection to make sure that the orignal accessTokenProvider that we registered is still registered before the redirect.
+        hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
+        assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
+        hubConnection.stop();
+        assertNull(beforeRedirectToken.get());
+        assertEquals("Bearer newToken", token.get());
+    }
 }