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

Client to Sever Streaming Java Edition (#8222)

This commit is contained in:
Mikael Mengistu 2019-03-07 09:34:28 -08:00 committed by GitHub
parent 5170c31b23
commit b1f828e161
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 346 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/HubConnection.java
View File

@ -3,6 +3,7 @@
package com.microsoft.signalr; package com.microsoft.signalr;
import java.lang.reflect.Array;
import java.util.*; import java.util.*;
import java.util.concurrent.*; import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
@ -49,6 +50,7 @@ public class HubConnection {
private long tickRate = 1000; private long tickRate = 1000;
private CompletableSubject handshakeResponseSubject; private CompletableSubject handshakeResponseSubject;
private long handshakeResponseTimeout = 15*1000; private long handshakeResponseTimeout = 15*1000;
private Map<String, Observable> streamMap = new ConcurrentHashMap<>();
private TransportEnum transportEnum = TransportEnum.ALL; private TransportEnum transportEnum = TransportEnum.ALL;
private final Logger logger = LoggerFactory.getLogger(HubConnection.class); private final Logger logger = LoggerFactory.getLogger(HubConnection.class);
@ -495,8 +497,54 @@ public class HubConnection {
throw new RuntimeException("The 'send' method cannot be called if the connection is not active."); throw new RuntimeException("The 'send' method cannot be called if the connection is not active.");
} }
InvocationMessage invocationMessage = new InvocationMessage(null, method, args); sendInvocationMessage(method, args);
}
private void sendInvocationMessage(String method, Object[] args) {
sendInvocationMessage(method, args, null, false);
}
private void sendInvocationMessage(String method, Object[] args, String id, Boolean isStreamInvocation) {
List<String> streamIds = new ArrayList<>();
args = checkUploadStream(args, streamIds);
InvocationMessage invocationMessage;
if (isStreamInvocation) {
invocationMessage = new StreamInvocationMessage(id, method, args, streamIds);
} else {
invocationMessage = new InvocationMessage(id, method, args, streamIds);
}
sendHubMessage(invocationMessage); sendHubMessage(invocationMessage);
launchStreams(streamIds);
}
void launchStreams(List<String> streamIds) {
if (streamMap.isEmpty()) {
return;
}
for (String streamId: streamIds) {
Observable observable = this.streamMap.get(streamId);
observable.subscribe(
(item) -> sendHubMessage(new StreamItem(streamId, item)),
(error) -> sendHubMessage(new CompletionMessage(streamId, null, error.toString())),
() -> sendHubMessage(new CompletionMessage(streamId, null, null)));
}
}
Object[] checkUploadStream(Object[] args, List<String> streamIds) {
List<Object> params = new ArrayList<>(Arrays.asList(args));
for (Object arg: args) {
if(arg instanceof Observable) {
params.remove(arg);
Observable stream = (Observable)arg;
String streamId = connectionState.getNextInvocationId();
streamIds.add(streamId);
this.streamMap.put(streamId, stream);
}
}
return params.toArray();
} }
/** /**
@ -515,7 +563,6 @@ public class HubConnection {
} }
String id = connectionState.getNextInvocationId(); String id = connectionState.getNextInvocationId();
InvocationMessage invocationMessage = new InvocationMessage(id, method, args);
SingleSubject<T> subject = SingleSubject.create(); SingleSubject<T> subject = SingleSubject.create();
InvocationRequest irq = new InvocationRequest(returnType, id); InvocationRequest irq = new InvocationRequest(returnType, id);
@ -535,8 +582,7 @@ public class HubConnection {
// Make sure the actual send is after setting up the callbacks otherwise there is a race // Make sure the actual send is after setting up the callbacks otherwise there is a race
// where the map doesn't have the callbacks yet when the response is returned // where the map doesn't have the callbacks yet when the response is returned
sendHubMessage(invocationMessage); sendInvocationMessage(method, args, id, false);
return subject; return subject;
} }
@ -553,7 +599,6 @@ public class HubConnection {
public <T> Observable<T> stream(Class<T> returnType, String method, Object ... args) { public <T> Observable<T> stream(Class<T> returnType, String method, Object ... args) {
String invocationId = connectionState.getNextInvocationId(); String invocationId = connectionState.getNextInvocationId();
AtomicInteger subscriptionCount = new AtomicInteger(); AtomicInteger subscriptionCount = new AtomicInteger();
StreamInvocationMessage streamInvocationMessage = new StreamInvocationMessage(invocationId, method, args);
InvocationRequest irq = new InvocationRequest(returnType, invocationId); InvocationRequest irq = new InvocationRequest(returnType, invocationId);
connectionState.addInvocation(irq); connectionState.addInvocation(irq);
ReplaySubject<T> subject = ReplaySubject.create(); ReplaySubject<T> subject = ReplaySubject.create();
@ -569,9 +614,8 @@ public class HubConnection {
}, error -> subject.onError(error), }, error -> subject.onError(error),
() -> subject.onComplete()); () -> subject.onComplete());
sendHubMessage(streamInvocationMessage);
Observable<T> observable = subject.doOnSubscribe((subscriber) -> subscriptionCount.incrementAndGet()); Observable<T> observable = subject.doOnSubscribe((subscriber) -> subscriptionCount.incrementAndGet());
sendInvocationMessage(method, args, invocationId, true);
return observable.doOnDispose(() -> { return observable.doOnDispose(() -> {
if (subscriptionCount.decrementAndGet() == 0) { if (subscriptionCount.decrementAndGet() == 0) {
CancelInvocationMessage cancelInvocationMessage = new CancelInvocationMessage(invocationId); CancelInvocationMessage cancelInvocationMessage = new CancelInvocationMessage(invocationId);
@ -591,8 +635,8 @@ public class HubConnection {
} else { } else {
logger.debug("Sending {} message.", message.getMessageType().name()); logger.debug("Sending {} message.", message.getMessageType().name());
} }
transport.send(serializedMessage).subscribeWith(CompletableSubject.create());
transport.send(serializedMessage).subscribeWith(CompletableSubject.create());
resetKeepAlive(); resetKeepAlive();
} }

12
src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/InvocationMessage.java
View File

@ -3,16 +3,26 @@
package com.microsoft.signalr; package com.microsoft.signalr;
import java.util.Collection;
class InvocationMessage extends HubMessage { class InvocationMessage extends HubMessage {
private final int type = HubMessageType.INVOCATION.value; int type = HubMessageType.INVOCATION.value;
private final String invocationId; private final String invocationId;
private final String target; private final String target;
private final Object[] arguments; private final Object[] arguments;
private Collection<String> streamIds;
public InvocationMessage(String invocationId, String target, Object[] args) { public InvocationMessage(String invocationId, String target, Object[] args) {
this(invocationId, target, args, null);
}
public InvocationMessage(String invocationId, String target, Object[] args, Collection<String> streamIds) {
this.invocationId = invocationId; this.invocationId = invocationId;
this.target = target; this.target = target;
this.arguments = args; this.arguments = args;
if(streamIds != null && !streamIds.isEmpty()) {
this.streamIds = streamIds;
}
} }
public String getInvocationId() { public String getInvocationId() {

26
src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/StreamInvocationMessage.java
View File

@ -3,28 +3,18 @@
package com.microsoft.signalr; package com.microsoft.signalr;
final class StreamInvocationMessage extends HubMessage { import java.util.Collection;
private final int type = HubMessageType.STREAM_INVOCATION.value;
private final String invocationId; final class StreamInvocationMessage extends InvocationMessage {
private final String target;
private final Object[] arguments;
public StreamInvocationMessage(String invocationId, String target, Object[] args) { public StreamInvocationMessage(String invocationId, String target, Object[] args) {
this.invocationId = invocationId; super(invocationId, target, args);
this.target = target; super.type = HubMessageType.STREAM_INVOCATION.value;
this.arguments = args;
} }
public String getInvocationId() { public StreamInvocationMessage(String invocationId, String target, Object[] args, Collection<String> streamIds) {
return invocationId; super(invocationId, target, args, streamIds);
} super.type = HubMessageType.STREAM_INVOCATION.value;
public String getTarget() {
return target;
}
public Object[] getArguments() {
return arguments;
} }
@Override @Override

277
src/SignalR/clients/java/signalr/src/test/java/com/microsoft/signalr/HubConnectionTest.java
View File

@ -14,7 +14,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import io.reactivex.subjects.CompletableSubject; import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.ReplaySubject;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import io.reactivex.Observable; import io.reactivex.Observable;
@ -371,6 +372,279 @@ class HubConnectionTest {
assertEquals(Double.valueOf(24), value.get()); assertEquals(Double.valueOf(24), value.get());
} }
@Test
public void checkStreamUploadSingleItemThroughSend() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream);
stream.onNext("FirstItem");
String[] messages = mockTransport.getSentMessages();
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"FirstItem\"}\u001E", messages[2]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[3]);
}
@Test
public void checkStreamUploadMultipleStreamsThroughSend() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> firstStream = ReplaySubject.create();
ReplaySubject<String> secondStream = ReplaySubject.create();
hubConnection.send("UploadStream", firstStream, secondStream);
firstStream.onNext("First Stream 1");
secondStream.onNext("Second Stream 1");
String[] messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"First Stream 1\"}\u001E", messages[2]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"Second Stream 1\"}\u001E", messages[3]);
firstStream.onComplete();
secondStream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(6, messages.length);
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[4]);
assertEquals("{\"type\":3,\"invocationId\":\"2\"}\u001E", messages[5]);
}
@Test
public void checkStreamUploadThroughSendWithArgs() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream, 12);
stream.onNext("FirstItem");
String[] messages = mockTransport.getSentMessages();
assertEquals("{\"type\":1,\"target\":\"UploadStream\",\"arguments\":[12],\"streamIds\":[\"1\"]}\u001E", messages[1]);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"FirstItem\"}\u001E", messages[2]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[3]);
}
@Test
public void useSameSubjectMultipleTimes() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream, stream);
stream.onNext("FirstItem");
String[] messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
assertEquals("{\"type\":1,\"target\":\"UploadStream\",\"arguments\":[],\"streamIds\":[\"1\",\"2\"]}\u001E", messages[1]);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"FirstItem\"}\u001E", messages[2]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"FirstItem\"}\u001E", messages[3]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(6, messages.length);
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[4]);
assertEquals("{\"type\":3,\"invocationId\":\"2\"}\u001E", messages[5]);
}
@Test
public void checkStreamUploadSingleItemThroughInvoke() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.invoke(String.class, "UploadStream", stream);
stream.onNext("FirstItem");
String[] messages = mockTransport.getSentMessages();
assertEquals(3, messages.length);
assertEquals("{\"type\":1,\"invocationId\":\"1\",\"target\":\"UploadStream\",\"arguments\":[],\"streamIds\":[\"2\"]}\u001E", messages[1]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"FirstItem\"}\u001E", messages[2]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals("{\"type\":3,\"invocationId\":\"2\"}\u001E", messages[3]);
}
@Test
public void checkStreamUploadSingleItemThroughStream() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.stream(String.class, "UploadStream", stream);
stream.onNext("FirstItem");
String[] messages = mockTransport.getSentMessages();
assertEquals(3, messages.length);
assertEquals("{\"type\":4,\"invocationId\":\"1\",\"target\":\"UploadStream\",\"arguments\":[],\"streamIds\":[\"2\"]}\u001E", messages[1]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"FirstItem\"}\u001E", messages[2]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
assertEquals("{\"type\":3,\"invocationId\":\"2\"}\u001E", messages[3]);
}
@Test
public void useSameSubjectInMutlipleStreamsFromDifferentMethods() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream);
hubConnection.invoke(String.class, "UploadStream", stream);
hubConnection.stream(String.class, "UploadStream", stream);
String[] messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
assertEquals("{\"type\":1,\"target\":\"UploadStream\",\"arguments\":[],\"streamIds\":[\"1\"]}\u001E", messages[1]);
assertEquals("{\"type\":1,\"invocationId\":\"2\",\"target\":\"UploadStream\",\"arguments\":[],\"streamIds\":[\"3\"]}\u001E", messages[2]);
assertEquals("{\"type\":4,\"invocationId\":\"4\",\"target\":\"UploadStream\",\"arguments\":[],\"streamIds\":[\"5\"]}\u001E", messages[3]);
stream.onNext("FirstItem");
messages = mockTransport.getSentMessages();
assertEquals(7, messages.length);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"FirstItem\"}\u001E", messages[4]);
assertEquals("{\"type\":2,\"invocationId\":\"3\",\"item\":\"FirstItem\"}\u001E", messages[5]);
assertEquals("{\"type\":2,\"invocationId\":\"5\",\"item\":\"FirstItem\"}\u001E", messages[6]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(10, messages.length);
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[7]);
assertEquals("{\"type\":3,\"invocationId\":\"3\"}\u001E", messages[8]);
assertEquals("{\"type\":3,\"invocationId\":\"5\"}\u001E", messages[9]);
}
@Test
public void streamUploadCallOnError() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream);
stream.onNext("FirstItem");
stream.onError(new RuntimeException("onError called"));
String[] messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"FirstItem\"}\u001E", messages[2]);
assertEquals("{\"type\":3,\"invocationId\":\"1\",\"error\":\"java.lang.RuntimeException: onError called\"}\u001E", messages[3]);
// onComplete doesn't send a completion message after onError.
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
}
@Test
public void checkStreamUploadMultipleItemsThroughSend() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.send("UploadStream", stream);
stream.onNext("FirstItem");
stream.onNext("SecondItem");
stream.onNext("ThirdItem");
String[] messages = mockTransport.getSentMessages();
assertEquals(5, messages.length);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"FirstItem\"}\u001E", messages[2]);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"SecondItem\"}\u001E", messages[3]);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"ThirdItem\"}\u001E", messages[4]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(6, messages.length);
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[5]);
}
@Test
public void checkStreamUploadMultipleItemsThroughInvoke() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
ReplaySubject<String> stream = ReplaySubject.create();
hubConnection.invoke(String.class, "UploadStream", stream);
stream.onNext("FirstItem");
stream.onNext("SecondItem");
String[] messages = mockTransport.getSentMessages();
assertEquals(4, messages.length);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"FirstItem\"}\u001E", messages[2]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"SecondItem\"}\u001E", messages[3]);
stream.onComplete();
messages = mockTransport.getSentMessages();
assertEquals(5, messages.length);
assertEquals("{\"type\":3,\"invocationId\":\"2\"}\u001E", messages[4]);
}
@Test
public void canStartAndStopMultipleStreams() {
MockTransport mockTransport = new MockTransport();
HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
PublishSubject<String> streamOne = PublishSubject.create();
PublishSubject<String> streamTwo = PublishSubject.create();
hubConnection.send("UploadStream", streamOne);
hubConnection.send("UploadStream", streamTwo);
streamOne.onNext("Stream One First Item");
streamTwo.onNext("Stream Two First Item");
streamOne.onNext("Stream One Second Item");
streamTwo.onNext("Stream Two Second Item");
streamOne.onComplete();
streamTwo.onComplete();
String[] messages = mockTransport.getSentMessages();
// Handshake message + 2 calls to send + 4 calls to onNext + 2 calls to onComplete = 9
assertEquals(9, messages.length);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"Stream One First Item\"}\u001E", messages[3]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"Stream Two First Item\"}\u001E", messages[4]);
assertEquals("{\"type\":2,\"invocationId\":\"1\",\"item\":\"Stream One Second Item\"}\u001E", messages[5]);
assertEquals("{\"type\":2,\"invocationId\":\"2\",\"item\":\"Stream Two Second Item\"}\u001E", messages[6]);
assertEquals("{\"type\":3,\"invocationId\":\"1\"}\u001E", messages[7]);
assertEquals("{\"type\":3,\"invocationId\":\"2\"}\u001E", messages[8]);
}
@Test @Test
public void checkStreamSingleItem() { public void checkStreamSingleItem() {
MockTransport mockTransport = new MockTransport(); MockTransport mockTransport = new MockTransport();
@ -426,7 +700,6 @@ class HubConnectionTest {
assertEquals("First", result.timeout(1000, TimeUnit.MILLISECONDS).blockingFirst()); assertEquals("First", result.timeout(1000, TimeUnit.MILLISECONDS).blockingFirst());
assertEquals("COMPLETED", result.timeout(1000, TimeUnit.MILLISECONDS).blockingLast()); assertEquals("COMPLETED", result.timeout(1000, TimeUnit.MILLISECONDS).blockingLast());
} }
@Test @Test