Skip to content

Commit 9b62ba4

Browse files
BrennanConroyBrennanConroy
committed
TS
1 parent 7810541 commit 9b62ba4

File tree

24 files changed

+583
-302
lines changed

24 files changed

+583
-302
lines changed

‎src/SignalR/benchmarks/Microsoft.AspNetCore.SignalR.Microbenchmarks/DefaultHubDispatcherBenchmark.cs

Lines changed: 83 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -38,8 +38,8 @@ public void GlobalSetup()
3838
_dispatcher = new DefaultHubDispatcher<TestHub>(
3939
serviceScopeFactory,
4040
new HubContext<TestHub>(new DefaultHubLifetimeManager<TestHub>(NullLogger<DefaultHubLifetimeManager<TestHub>>.Instance)),
41-
Options.Create(new HubOptions<TestHub>()),
42-
Options.Create(new HubOptions()),
41+
Options.Create(new HubOptions<TestHub>() { EnableDetailedErrors = true }),
42+
Options.Create(new HubOptions() { EnableDetailedErrors = true }),
4343
new Logger<DefaultHubDispatcher<TestHub>>(NullLoggerFactory.Instance));
4444

4545
var pair = DuplexPipe.CreateConnectionPair(PipeOptions.Default, PipeOptions.Default);
@@ -163,61 +163,71 @@ public ChannelReader<int> StreamChannelReaderCount(int count)
163163

164164
return channel.Reader;
165165
}
166-
}
167-
168-
[Benchmark]
169-
public Task Invocation()
170-
{
171-
return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "Invocation", Array.Empty<object>()));
172-
}
173-
174-
[Benchmark]
175-
public Task InvocationAsync()
176-
{
177-
return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationAsync", Array.Empty<object>()));
178-
}
179-
180-
[Benchmark]
181-
public Task InvocationReturnValue()
182-
{
183-
return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationReturnValue", Array.Empty<object>()));
184-
}
185-
186-
[Benchmark]
187-
public Task InvocationReturnAsync()
188-
{
189-
return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationReturnAsync", Array.Empty<object>()));
190-
}
191-
192-
[Benchmark]
193-
public Task InvocationValueTaskAsync()
194-
{
195-
return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationValueTaskAsync", Array.Empty<object>()));
196-
}
197166

198-
[Benchmark]
199-
public Task StreamChannelReader()
200-
{
201-
return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReader", Array.Empty<object>()));
202-
}
203-
204-
[Benchmark]
205-
public Task StreamChannelReaderAsync()
206-
{
207-
return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderAsync", Array.Empty<object>()));
208-
}
209-
210-
[Benchmark]
211-
public Task StreamChannelReaderValueTaskAsync()
212-
{
213-
return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderValueTaskAsync", Array.Empty<object>()));
167+
public async Task UploadStream(ChannelReader<string> channelReader)
168+
{
169+
while (await channelReader.WaitToReadAsync())
170+
{
171+
while (channelReader.TryRead(out var item))
172+
{
173+
}
174+
}
175+
}
214176
}
215177

216-
[Benchmark]
217-
public Task StreamChannelReaderCount_Zero()
218-
{
219-
return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderCount", new object[] { 0 }));
220-
}
178+
//[Benchmark]
179+
//public Task Invocation()
180+
//{
181+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "Invocation", Array.Empty<object>()));
182+
//}
183+
184+
//[Benchmark]
185+
//public Task InvocationAsync()
186+
//{
187+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationAsync", Array.Empty<object>()));
188+
//}
189+
190+
//[Benchmark]
191+
//public Task InvocationReturnValue()
192+
//{
193+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationReturnValue", Array.Empty<object>()));
194+
//}
195+
196+
//[Benchmark]
197+
//public Task InvocationReturnAsync()
198+
//{
199+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationReturnAsync", Array.Empty<object>()));
200+
//}
201+
202+
//[Benchmark]
203+
//public Task InvocationValueTaskAsync()
204+
//{
205+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", "InvocationValueTaskAsync", Array.Empty<object>()));
206+
//}
207+
208+
//[Benchmark]
209+
//public Task StreamChannelReader()
210+
//{
211+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReader", Array.Empty<object>()));
212+
//}
213+
214+
//[Benchmark]
215+
//public Task StreamChannelReaderAsync()
216+
//{
217+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderAsync", Array.Empty<object>()));
218+
//}
219+
220+
//[Benchmark]
221+
//public Task StreamChannelReaderValueTaskAsync()
222+
//{
223+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderValueTaskAsync", Array.Empty<object>()));
224+
//}
225+
226+
//[Benchmark]
227+
//public Task StreamChannelReaderCount_Zero()
228+
//{
229+
// return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderCount", new object[] { 0 }));
230+
//}
221231

222232
[Benchmark]
223233
public Task StreamChannelReaderCount_One()
@@ -230,5 +240,24 @@ public Task StreamChannelReaderCount_Thousand()
230240
{
231241
return _dispatcher.DispatchMessageAsync(_connectionContext, new StreamInvocationMessage("123", "StreamChannelReaderCount", new object[] { 1000 }));
232242
}
243+
244+
//[Benchmark]
245+
//public async Task UploadStream_One()
246+
//{
247+
// await _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", nameof(TestHub.UploadStream), Array.Empty<object>(), streamIds: new string[] { "1" }));
248+
// await _dispatcher.DispatchMessageAsync(_connectionContext, new StreamItemMessage("1", "test"));
249+
// //await _dispatcher.DispatchMessageAsync(_connectionContext, CompletionMessage.Empty("1"));
250+
//}
251+
252+
//[Benchmark]
253+
//public async Task UploadStream_Thousand()
254+
//{
255+
// await _dispatcher.DispatchMessageAsync(_connectionContext, new InvocationMessage("123", nameof(TestHub.UploadStream), Array.Empty<object>(), streamIds: new string[] { "1" }));
256+
// for (var i = 0; i < 1000; ++i)
257+
// {
258+
// await _dispatcher.DispatchMessageAsync(_connectionContext, new StreamItemMessage("1", "test"));
259+
// }
260+
// //await _dispatcher.DispatchMessageAsync(_connectionContext, CompletionMessage.Empty("1"));
261+
//}
233262
}
234263
}

‎src/SignalR/clients/ts/signalr-protocol-msgpack/src/MessagePackHubProtocol.ts

Lines changed: 32 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import { Buffer } from "buffer";
55
import * as msgpack5 from "msgpack5";
66

7-
import { CompletionMessage, HubMessage, IHubProtocol, ILogger, InvocationMessage, LogLevel, MessageHeaders, MessageType, NullLogger, StreamCompleteMessage, StreamDataMessage, StreamInvocationMessage, StreamItemMessage, TransferFormat } from "@aspnet/signalr";
7+
import { CompletionMessage, HubMessage, IHubProtocol, ILogger, InvocationMessage, LogLevel, MessageHeaders, MessageType, NullLogger, StreamInvocationMessage, StreamItemMessage, TransferFormat } from "@aspnet/signalr";
88

99
import { BinaryMessageFormat } from "./BinaryMessageFormat";
1010
import { isArrayBuffer } from "./Utils";
@@ -25,6 +25,10 @@ export class MessagePackHubProtocol implements IHubProtocol {
2525
/** The TransferFormat of the protocol. */
2626
public readonly transferFormat: TransferFormat = TransferFormat.Binary;
2727

28+
private readonly errorResult = 1;
29+
private readonly voidResult = 2;
30+
private readonly nonVoidResult = 3;
31+
2832
/** Creates an array of HubMessage objects from the specified serialized representation.
2933
*
3034
* @param {ArrayBuffer | Buffer} input An ArrayBuffer containing the serialized representation.
@@ -65,15 +69,12 @@ export class MessagePackHubProtocol implements IHubProtocol {
6569
return this.writeInvocation(message as InvocationMessage);
6670
case MessageType.StreamInvocation:
6771
return this.writeStreamInvocation(message as StreamInvocationMessage);
68-
case MessageType.StreamData:
69-
return this.writeStreamData(message as StreamDataMessage);
7072
case MessageType.StreamItem:
73+
return this.writeStreamItem(message as StreamItemMessage);
7174
case MessageType.Completion:
72-
throw new Error(`Writing messages of type '${message.type}' is not supported.`);
75+
return this.writeCompletion(message as CompletionMessage);
7376
case MessageType.Ping:
7477
return BinaryMessageFormat.write(SERIALIZED_PING_MESSAGE);
75-
case MessageType.StreamComplete:
76-
return this.writeStreamComplete(message as StreamCompleteMessage);
7778
default:
7879
throw new Error("Invalid message type.");
7980
}
@@ -147,13 +148,15 @@ export class MessagePackHubProtocol implements IHubProtocol {
147148
arguments: properties[4],
148149
headers,
149150
invocationId,
151+
streamIds: [],
150152
target: properties[3] as string,
151153
type: MessageType.Invocation,
152154
};
153155
} else {
154156
return {
155157
arguments: properties[4],
156158
headers,
159+
streamIds: [],
157160
target: properties[3],
158161
type: MessageType.Invocation,
159162
};
@@ -181,24 +184,20 @@ export class MessagePackHubProtocol implements IHubProtocol {
181184
throw new Error("Invalid payload for Completion message.");
182185
}
183186

184-
const errorResult = 1;
185-
const voidResult = 2;
186-
const nonVoidResult = 3;
187-
188187
const resultKind = properties[3];
189188

190-
if (resultKind !== voidResult && properties.length < 5) {
189+
if (resultKind !== this.voidResult && properties.length < 5) {
191190
throw new Error("Invalid payload for Completion message.");
192191
}
193192

194193
let error: string | undefined;
195194
let result: any;
196195

197196
switch (resultKind) {
198-
case errorResult:
197+
case this.errorResult:
199198
error = properties[4];
200199
break;
201-
case nonVoidResult:
200+
case this.nonVoidResult:
202201
result = properties[4];
203202
break;
204203
}
@@ -217,31 +216,43 @@ export class MessagePackHubProtocol implements IHubProtocol {
217216
private writeInvocation(invocationMessage: InvocationMessage): ArrayBuffer {
218217
const msgpack = msgpack5();
219218
const payload = msgpack.encode([MessageType.Invocation, invocationMessage.headers || {}, invocationMessage.invocationId || null,
220-
invocationMessage.target, invocationMessage.arguments]);
219+
invocationMessage.target, invocationMessage.arguments, invocationMessage.streamIds]);
221220

222221
return BinaryMessageFormat.write(payload.slice());
223222
}
224223

225224
private writeStreamInvocation(streamInvocationMessage: StreamInvocationMessage): ArrayBuffer {
226225
const msgpack = msgpack5();
227226
const payload = msgpack.encode([MessageType.StreamInvocation, streamInvocationMessage.headers || {}, streamInvocationMessage.invocationId,
228-
streamInvocationMessage.target, streamInvocationMessage.arguments]);
227+
streamInvocationMessage.target, streamInvocationMessage.arguments, streamInvocationMessage.streamIds]);
229228

230229
return BinaryMessageFormat.write(payload.slice());
231230
}
232231

233-
private writeStreamData(streamDataMessage: StreamDataMessage): ArrayBuffer {
232+
private writeStreamItem(streamItemMessage: StreamItemMessage): ArrayBuffer {
234233
const msgpack = msgpack5();
235-
const payload = msgpack.encode([MessageType.StreamData, streamDataMessage.streamId,
236-
streamDataMessage.item]);
234+
const payload = msgpack.encode([MessageType.StreamItem, streamItemMessage.headers || {}, streamItemMessage.invocationId,
235+
streamItemMessage.item]);
237236

238237
return BinaryMessageFormat.write(payload.slice());
239238
}
240239

241-
private writeStreamComplete(streamCompleteMessage: StreamCompleteMessage): ArrayBuffer {
240+
private writeCompletion(completionMessage: CompletionMessage): ArrayBuffer {
242241
const msgpack = msgpack5();
243-
const payload = msgpack.encode([MessageType.StreamComplete, streamCompleteMessage.streamId,
244-
streamCompleteMessage.error || null]);
242+
const resultKind = completionMessage.error ? this.errorResult : completionMessage.result ? this.nonVoidResult : this.voidResult;
243+
244+
let payload: any;
245+
switch (resultKind) {
246+
case this.errorResult:
247+
payload = msgpack.encode([MessageType.Completion, completionMessage.headers || {}, completionMessage.invocationId, resultKind, completionMessage.error]);
248+
break;
249+
case this.voidResult:
250+
payload = msgpack.encode([MessageType.Completion, completionMessage.headers || {}, completionMessage.invocationId, resultKind]);
251+
break;
252+
case this.nonVoidResult:
253+
payload = msgpack.encode([MessageType.Completion, completionMessage.headers || {}, completionMessage.invocationId, resultKind, completionMessage.result]);
254+
break;
255+
}
245256

246257
return BinaryMessageFormat.write(payload.slice());
247258
}

‎src/SignalR/clients/ts/signalr-protocol-msgpack/tests/MessagePackHubProtocol.test.ts

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ describe("MessagePackHubProtocol", () => {
99
const invocation = {
1010
arguments: [42, true, "test", ["x1", "y2"], null],
1111
headers: {},
12+
streamIds: [],
1213
target: "myMethod",
1314
type: MessageType.Invocation,
1415
} as InvocationMessage;
@@ -22,6 +23,7 @@ describe("MessagePackHubProtocol", () => {
2223
const invocation = {
2324
arguments: [new Date(Date.UTC(2018, 1, 1, 12, 34, 56))],
2425
headers: {},
26+
streamIds: [],
2527
target: "mymethod",
2628
type: MessageType.Invocation,
2729
} as InvocationMessage;
@@ -37,6 +39,7 @@ describe("MessagePackHubProtocol", () => {
3739
headers: {
3840
foo: "bar",
3941
},
42+
streamIds: [],
4043
target: "myMethod",
4144
type: MessageType.Invocation,
4245
} as InvocationMessage;
@@ -51,6 +54,7 @@ describe("MessagePackHubProtocol", () => {
5154
arguments: [42, true, "test", ["x1", "y2"], null],
5255
headers: {},
5356
invocationId: "123",
57+
streamIds: [],
5458
target: "myMethod",
5559
type: MessageType.Invocation,
5660
} as InvocationMessage;

0 commit comments

Comments
 (0)