huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Make Java client more reusable (#3029)

This commit is contained in:
BrennanConroy 2018-10-02 14:08:36 -07:00 committed by GitHub
parent 17e029f15d
commit 3ae2814c27
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 58 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
clients/java/signalr/src/main/java/com/microsoft/aspnet/signalr/HubConnection.java
View File

@ -16,7 +16,7 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer; import java.util.function.Consumer;
public class HubConnection { public class HubConnection {
private String url; private String baseUrl;
private Transport transport; private Transport transport;
private OnReceiveCallBack callback; private OnReceiveCallBack callback;
private CallbackMap handlers = new CallbackMap(); private CallbackMap handlers = new CallbackMap();
@ -41,7 +41,7 @@ public class HubConnection {
throw new IllegalArgumentException("A valid url is required."); throw new IllegalArgumentException("A valid url is required.");
} }
this.url = url; this.baseUrl = url;
this.protocol = new JsonHubProtocol(); this.protocol = new JsonHubProtocol();
if (logger != null) { if (logger != null) {
@ -126,7 +126,7 @@ public class HubConnection {
}; };
} }
private CompletableFuture<NegotiateResponse> handleNegotiate() throws IOException, InterruptedException, ExecutionException { private CompletableFuture<NegotiateResponse> handleNegotiate(String url) throws IOException, InterruptedException, ExecutionException {
HttpRequest request = new HttpRequest(); HttpRequest request = new HttpRequest();
request.setHeaders(this.headers); request.setHeaders(this.headers);
@ -142,22 +142,10 @@ public class HubConnection {
throw new RuntimeException(negotiateResponse.getError()); throw new RuntimeException(negotiateResponse.getError());
} }
if (negotiateResponse.getConnectionId() != null) {
if (url.contains("?")) {
url = url + "&id=" + negotiateResponse.getConnectionId();
} else {
url = url + "?id=" + negotiateResponse.getConnectionId();
}
}
if (negotiateResponse.getAccessToken() != null) { if (negotiateResponse.getAccessToken() != null) {
this.headers.put("Authorization", "Bearer " + negotiateResponse.getAccessToken()); this.headers.put("Authorization", "Bearer " + negotiateResponse.getAccessToken());
} }
if (negotiateResponse.getRedirectUrl() != null) {
this.url = negotiateResponse.getRedirectUrl();
}
return CompletableFuture.completedFuture(negotiateResponse); return CompletableFuture.completedFuture(negotiateResponse);
}); });
} }
@ -181,20 +169,14 @@ public class HubConnection {
return CompletableFuture.completedFuture(null); return CompletableFuture.completedFuture(null);
} }
CompletableFuture<NegotiateResponse> negotiate = null; CompletableFuture<String> negotiate = null;
if (!skipNegotiate) { if (!skipNegotiate) {
negotiate = startNegotiate(0); negotiate = startNegotiate(baseUrl, 0);
} else { } else {
negotiate = CompletableFuture.completedFuture(null); negotiate = CompletableFuture.completedFuture(baseUrl);
} }
return negotiate.thenCompose((response) -> { return negotiate.thenCompose((url) -> {
// If we didn't skip negotiate and got a null response then exit start because we
// are probably disconnected
if (response == null && !skipNegotiate) {
return CompletableFuture.completedFuture(null);
}
logger.log(LogLevel.Debug, "Starting HubConnection"); logger.log(LogLevel.Debug, "Starting HubConnection");
if (transport == null) { if (transport == null) {
transport = new WebSocketTransport(url, headers, httpClient, logger); transport = new WebSocketTransport(url, headers, httpClient, logger);
@ -225,12 +207,12 @@ public class HubConnection {
}); });
} }
private CompletableFuture<NegotiateResponse> startNegotiate(int negotiateAttempts) throws IOException, InterruptedException, ExecutionException { private CompletableFuture<String> startNegotiate(String url, int negotiateAttempts) throws IOException, InterruptedException, ExecutionException {
if (hubConnectionState != HubConnectionState.DISCONNECTED) { if (hubConnectionState != HubConnectionState.DISCONNECTED) {
return CompletableFuture.completedFuture(null); return CompletableFuture.completedFuture(null);
} }
return handleNegotiate().thenCompose((response) -> { return handleNegotiate(url).thenCompose((response) -> {
if (response.getRedirectUrl() != null && negotiateAttempts >= MAX_NEGOTIATE_ATTEMPTS) { if (response.getRedirectUrl() != null && negotiateAttempts >= MAX_NEGOTIATE_ATTEMPTS) {
throw new RuntimeException("Negotiate redirection limit exceeded."); throw new RuntimeException("Negotiate redirection limit exceeded.");
} }
@ -243,11 +225,21 @@ public class HubConnection {
throw new RuntimeException(e); throw new RuntimeException(e);
} }
} }
return CompletableFuture.completedFuture(response);
String finalUrl = url;
if (response.getConnectionId() != null) {
if (url.contains("?")) {
finalUrl = url + "&id=" + response.getConnectionId();
} else {
finalUrl = url + "?id=" + response.getConnectionId();
}
}
return CompletableFuture.completedFuture(finalUrl);
} }
try { try {
return startNegotiate(negotiateAttempts + 1); return startNegotiate(response.getRedirectUrl(), negotiateAttempts + 1);
} catch (IOException | InterruptedException | ExecutionException e) { } catch (IOException | InterruptedException | ExecutionException e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }

39
clients/java/signalr/src/test/java/com/microsoft/aspnet/signalr/HubConnectionTest.java
View File

@ -948,8 +948,7 @@ class HubConnectionTest {
} }
@Test @Test
public void negotiateRedirectIsFollowed() public void negotiateRedirectIsFollowed() throws Exception {
throws InterruptedException, ExecutionException, TimeoutException, Exception {
TestHttpClient client = new TestHttpClient().on("POST", "http://example.com/negotiate", TestHttpClient client = new TestHttpClient().on("POST", "http://example.com/negotiate",
(req) -> CompletableFuture.completedFuture(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\"}"))) (req) -> CompletableFuture.completedFuture(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\"}")))
.on("POST", "http://testexample.com/negotiate", .on("POST", "http://testexample.com/negotiate",
@ -966,4 +965,40 @@ class HubConnectionTest {
assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState()); assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
hubConnection.stop(); hubConnection.stop();
} }
@Test
public void hubConnectionCanBeStartedAfterBeingStopped() throws Exception {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = new HubConnection("http://example.com", mockTransport, new NullLogger(), true, new TestHttpClient());
hubConnection.start().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
hubConnection.stop().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
hubConnection.start().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
hubConnection.stop().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
}
@Test
public void hubConnectionCanBeStartedAfterBeingStoppedAndRedirected() throws Exception {
MockTransport mockTransport = new MockTransport();
TestHttpClient client = new TestHttpClient()
.on("POST", "http://example.com/negotiate", (req) -> CompletableFuture
.completedFuture(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\"}")))
.on("POST", "http://testexample.com/negotiate", (req) -> CompletableFuture
.completedFuture(new HttpResponse(200, "", "{\"connectionId\":\"bVOiRPG8-6YiJ6d7ZcTOVQ\",\""
+ "availableTransports\":[{\"transport\":\"WebSockets\",\"transferFormats\":[\"Text\",\"Binary\"]}]}")));
HubConnection hubConnection = new HubConnection("http://example.com", mockTransport, new NullLogger(), false, client);
hubConnection.start().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
hubConnection.stop().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
hubConnection.start().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
hubConnection.stop().get(1000, TimeUnit.MILLISECONDS);
assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
}
} }