[TS Client] Test message size (#15099)

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

@@ -221,16 +221,28 @@ export class MessagePackHubProtocol implements IHubProtocol {
 
     private writeInvocation(invocationMessage: InvocationMessage): ArrayBuffer {
         const msgpack = msgpack5();
-        const payload = msgpack.encode([MessageType.Invocation, invocationMessage.headers || {}, invocationMessage.invocationId || null,
+        let payload: any;
-        invocationMessage.target, invocationMessage.arguments, invocationMessage.streamIds]);
+        if (invocationMessage.streamIds) {
+            payload = msgpack.encode([MessageType.Invocation, invocationMessage.headers || {}, invocationMessage.invocationId || null,
+            invocationMessage.target, invocationMessage.arguments, invocationMessage.streamIds]);
+        } else {
+            payload = msgpack.encode([MessageType.Invocation, invocationMessage.headers || {}, invocationMessage.invocationId || null,
+                invocationMessage.target, invocationMessage.arguments]);
+        }
 
         return BinaryMessageFormat.write(payload.slice());
     }
 
     private writeStreamInvocation(streamInvocationMessage: StreamInvocationMessage): ArrayBuffer {
         const msgpack = msgpack5();
-        const payload = msgpack.encode([MessageType.StreamInvocation, streamInvocationMessage.headers || {}, streamInvocationMessage.invocationId,
+        let payload: any;
-        streamInvocationMessage.target, streamInvocationMessage.arguments, streamInvocationMessage.streamIds]);
+        if (streamInvocationMessage.streamIds) {
+            payload = msgpack.encode([MessageType.StreamInvocation, streamInvocationMessage.headers || {}, streamInvocationMessage.invocationId,
+            streamInvocationMessage.target, streamInvocationMessage.arguments, streamInvocationMessage.streamIds]);
+        } else {
+            payload = msgpack.encode([MessageType.StreamInvocation, streamInvocationMessage.headers || {}, streamInvocationMessage.invocationId,
+                streamInvocationMessage.target, streamInvocationMessage.arguments]);
+        }
 
         return BinaryMessageFormat.write(payload.slice());
     }

src/SignalR/clients/ts/signalr/src/HubConnection.ts

@@ -818,23 +818,40 @@ export class HubConnection {
 
     private createInvocation(methodName: string, args: any[], nonblocking: boolean, streamIds: string[]): InvocationMessage {
         if (nonblocking) {
-            return {
+            if (streamIds.length !== 0) {
-                arguments: args,
+                return {
-                streamIds,
+                    arguments: args,
-                target: methodName,
+                    streamIds,
-                type: MessageType.Invocation,
+                    target: methodName,
-            };
+                    type: MessageType.Invocation,
+                };
+            } else {
+                return {
+                    arguments: args,
+                    target: methodName,
+                    type: MessageType.Invocation,
+                };
+            }
         } else {
             const invocationId = this.invocationId;
             this.invocationId++;
 
-            return {
+            if (streamIds.length !== 0) {
-                arguments: args,
+                return {
-                invocationId: invocationId.toString(),
+                    arguments: args,
-                streamIds,
+                    invocationId: invocationId.toString(),
-                target: methodName,
+                    streamIds,
-                type: MessageType.Invocation,
+                    target: methodName,
-            };
+                    type: MessageType.Invocation,
+                };
+            } else {
+                return {
+                    arguments: args,
+                    invocationId: invocationId.toString(),
+                    target: methodName,
+                    type: MessageType.Invocation,
+                };
+            }
         }
     }
 
@@ -903,13 +920,22 @@ export class HubConnection {
         const invocationId = this.invocationId;
         this.invocationId++;
 
-        return {
+        if (streamIds.length !== 0) {
-            arguments: args,
+            return {
-            invocationId: invocationId.toString(),
+                arguments: args,
-            streamIds,
+                invocationId: invocationId.toString(),
-            target: methodName,
+                streamIds,
-            type: MessageType.StreamInvocation,
+                target: methodName,
-        };
+                type: MessageType.StreamInvocation,
+            };
+        } else {
+            return {
+                arguments: args,
+                invocationId: invocationId.toString(),
+                target: methodName,
+                type: MessageType.StreamInvocation,
+            };
+        }
     }
 
     private createCancelInvocation(id: string): CancelInvocationMessage {

src/SignalR/clients/ts/signalr/src/IHubProtocol.ts

@@ -65,7 +65,7 @@ export interface InvocationMessage extends HubInvocationMessage {
     /** The target method arguments. */
     readonly arguments: any[];
     /** The target methods stream IDs. */
-    readonly streamIds: string[];
+    readonly streamIds?: string[];
 }
 
 /** A hub message representing a streaming invocation. */
@@ -80,7 +80,7 @@ export interface StreamInvocationMessage extends HubInvocationMessage {
     /** The target method arguments. */
     readonly arguments: any[];
     /** The target methods stream IDs. */
-    readonly streamIds: string[];
+    readonly streamIds?: string[];
 }
 
 /** A hub message representing a single item produced as part of a result stream. */

src/SignalR/clients/ts/signalr/tests/HubConnection.test.ts

@@ -184,7 +184,6 @@ describe("HubConnection", () => {
                             "arg",
                             42,
                         ],
-                        streamIds: [],
                         target: "testMethod",
                         type: MessageType.Invocation,
                     });
@@ -213,7 +212,6 @@ describe("HubConnection", () => {
                             "arg",
                             null,
                         ],
-                        streamIds: [],
                         target: "testMethod",
                         type: MessageType.Invocation,
                     });
@@ -245,7 +243,6 @@ describe("HubConnection", () => {
                             42,
                         ],
                         invocationId: connection.lastInvocationId,
-                        streamIds: [],
                         target: "testMethod",
                         type: MessageType.Invocation,
                     });
@@ -998,7 +995,6 @@ describe("HubConnection", () => {
                             42,
                         ],
                         invocationId: connection.lastInvocationId,
-                        streamIds: [],
                         target: "testStream",
                         type: MessageType.StreamInvocation,
                     });

src/SignalR/clients/ts/signalr/tests/MessageSize.test.ts

@@ -0,0 +1,175 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
+
+import { HubConnection } from "../src/HubConnection";
+import { IConnection } from "../src/IConnection";
+import { IHubProtocol, MessageType } from "../src/IHubProtocol";
+import { ILogger } from "../src/ILogger";
+import { JsonHubProtocol } from "../src/JsonHubProtocol";
+import { NullLogger } from "../src/Loggers";
+import { Subject } from "../src/Subject";
+import { VerifyLogger } from "./Common";
+import { TestConnection } from "./TestConnection";
+import { delayUntil, registerUnhandledRejectionHandler } from "./Utils";
+
+registerUnhandledRejectionHandler();
+
+function createHubConnection(connection: IConnection, logger?: ILogger | null, protocol?: IHubProtocol | null) {
+    return HubConnection.create(connection, logger || NullLogger.instance, protocol || new JsonHubProtocol());
+}
+
+// These tests check that the message size doesn't change without us being aware of it and making a conscious decision to increase the size
+
+describe("Message size", () => {
+    it("send invocation", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                // We don't actually care to wait for the send.
+                // tslint:disable-next-line:no-floating-promises
+                hubConnection.send("target", 1)
+                    .catch((_) => { }); // Suppress exception and unhandled promise rejection warning.
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(1);
+                expect(connection.parsedSentData[0].type).toEqual(MessageType.Invocation);
+                expect((connection.sentData[0] as string).length).toEqual(44);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+
+    it("invoke invocation", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                // We don't actually care to wait for the invoke.
+                // tslint:disable-next-line:no-floating-promises
+                hubConnection.invoke("target", 1)
+                    .catch((_) => { }); // Suppress exception and unhandled promise rejection warning.
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(1);
+                expect(connection.parsedSentData[0].type).toEqual(MessageType.Invocation);
+                expect((connection.sentData[0] as string).length).toEqual(63);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+
+    it("stream invocation", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                hubConnection.stream("target", 1);
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(1);
+                expect(connection.parsedSentData[0].type).toEqual(MessageType.StreamInvocation);
+                expect((connection.sentData[0] as string).length).toEqual(63);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+
+    it("upload invocation", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                // We don't actually care to wait for the invoke.
+                // tslint:disable-next-line:no-floating-promises
+                hubConnection.invoke("target", 1, new Subject())
+                    .catch((_) => { }); // Suppress exception and unhandled promise rejection warning.
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(1);
+                expect(connection.parsedSentData[0].type).toEqual(MessageType.Invocation);
+                expect((connection.sentData[0] as string).length).toEqual(81);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+
+    it("upload stream invocation", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                hubConnection.stream("target", 1, new Subject());
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(1);
+                expect(connection.parsedSentData[0].type).toEqual(MessageType.StreamInvocation);
+                expect((connection.sentData[0] as string).length).toEqual(81);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+
+    it("completion message", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                const subject = new Subject();
+                hubConnection.stream("target", 1, subject);
+                subject.complete();
+
+                await delayUntil(1000, () => connection.sentData.length === 2);
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(2);
+                expect(connection.parsedSentData[1].type).toEqual(MessageType.Completion);
+                expect((connection.sentData[1] as string).length).toEqual(29);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+
+    it("cancel message", async () => {
+        await VerifyLogger.run(async (logger) => {
+            const connection = new TestConnection();
+
+            const hubConnection = createHubConnection(connection, logger);
+            try {
+                hubConnection.stream("target", 1).subscribe({
+                    complete: () => {},
+                    error: () => {},
+                    next: () => {},
+                }).dispose();
+
+                await delayUntil(1000, () => connection.sentData.length === 2);
+
+                // Verify the message is sent
+                expect(connection.sentData.length).toBe(2);
+                expect(connection.parsedSentData[1].type).toEqual(MessageType.CancelInvocation);
+                expect((connection.sentData[1] as string).length).toEqual(29);
+            } finally {
+                // Close the connection
+                await hubConnection.stop();
+            }
+        });
+    });
+});

src/SignalR/clients/ts/signalr/tests/TestConnection.ts

@@ -13,6 +13,7 @@ export class TestConnection implements IConnection {
     public onclose: ((error?: Error) => void) | null;
 
     public sentData: any[];
+    public parsedSentData: any[];
     public lastInvocationId: string | null;
 
     private autoHandshake: boolean | null;
@@ -21,6 +22,7 @@ export class TestConnection implements IConnection {
         this.onreceive = null;
         this.onclose = null;
         this.sentData = [];
+        this.parsedSentData = [];
         this.lastInvocationId = null;
         this.autoHandshake = autoHandshake;
         this.baseUrl = "http://example.com";
@@ -43,8 +45,10 @@ export class TestConnection implements IConnection {
         }
         if (this.sentData) {
             this.sentData.push(invocation);
+            this.parsedSentData.push(parsedInvocation);
         } else {
             this.sentData = [invocation];
+            this.parsedSentData = [parsedInvocation];
         }
         return Promise.resolve();
     }

src/SignalR/common/SignalR.Common/test/Internal/Protocol/JsonHubProtocolTestsBase.cs

@@ -342,6 +342,7 @@ namespace Microsoft.AspNetCore.SignalR.Common.Tests.Internal.Protocol
 
         [Theory]
         [MemberData(nameof(MessageSizeDataNames))]
+        // These tests check that the message size doesn't change without us being aware of it and making a conscious decision to increase the size
         public void VerifyMessageSize(string testDataName)
         {
             var testData = MessageSizeData[testDataName];

src/SignalR/common/SignalR.Common/test/Internal/Protocol/MessagePackHubProtocolTestBase.cs

@@ -375,6 +375,57 @@ namespace Microsoft.AspNetCore.SignalR.Common.Tests.Internal.Protocol
             Assert.Null(message);
         }
 
+        public static IDictionary<string, MessageSizeTestData> MessageSizeData => new[]
+        {
+            new MessageSizeTestData("InvocationMessage_WithoutInvocationId", new InvocationMessage("Target", new object[] { 1 }), 15),
+            new MessageSizeTestData("InvocationMessage_WithInvocationId", new InvocationMessage("1", "Target", new object[] { 1 }), 16),
+            new MessageSizeTestData("InvocationMessage_WithInvocationIdAndStreamId", new InvocationMessage("1", "Target", new object[] { 1 }, new string[] { "2" }), 18),
+
+            new MessageSizeTestData("CloseMessage_Empty", CloseMessage.Empty, 5),
+            new MessageSizeTestData("CloseMessage_WithError", new CloseMessage("error"), 10),
+
+            new MessageSizeTestData("StreamItemMessage_WithNullItem", new StreamItemMessage("1", null), 7),
+            new MessageSizeTestData("StreamItemMessage_WithItem", new StreamItemMessage("1", 1), 7),
+
+            new MessageSizeTestData("CompletionMessage_Empty", CompletionMessage.Empty("1"), 7),
+            new MessageSizeTestData("CompletionMessage_WithResult", CompletionMessage.WithResult("1", 1), 8),
+            new MessageSizeTestData("CompletionMessage_WithError", CompletionMessage.WithError("1", "error"), 13),
+
+            new MessageSizeTestData("StreamInvocationMessage", new StreamInvocationMessage("1", "target", Array.Empty<object>()), 15),
+            new MessageSizeTestData("StreamInvocationMessage_WithStreamId", new StreamInvocationMessage("1", "target", Array.Empty<object>(), new [] { "2" }), 17),
+
+            new MessageSizeTestData("CancelInvocationMessage", new CancelInvocationMessage("1"), 6),
+
+            new MessageSizeTestData("PingMessage", PingMessage.Instance, 3),
+        }.ToDictionary(t => t.Name);
+
+        public static IEnumerable<object[]> MessageSizeDataNames => MessageSizeData.Keys.Select(name => new object[] { name });
+
+        [Theory]
+        [MemberData(nameof(MessageSizeDataNames))]
+        // These tests check that the message size doesn't change without us being aware of it and making a conscious decision to increase the size
+        public void VerifyMessageSize(string testDataName)
+        {
+            var testData = MessageSizeData[testDataName];
+            Assert.Equal(testData.Size, Write(testData.Message).Length);
+        }
+
+        public class MessageSizeTestData
+        {
+            public string Name { get; }
+            public HubMessage Message { get; }
+            public int Size { get; }
+
+            public MessageSizeTestData(string name, HubMessage message, int size)
+            {
+                Name = name;
+                Message = message;
+                Size = size;
+            }
+
+            public override string ToString() => Name;
+        }
+
         protected byte ArrayBytes(int size)
         {
             Debug.Assert(size < 16, "Test code doesn't support array sizes greater than 15");

src/SignalR/common/SignalR.Common/test/Internal/Protocol/MessagePackHubProtocolTests.cs

@@ -203,55 +203,5 @@ namespace Microsoft.AspNetCore.SignalR.Common.Tests.Internal.Protocol
 
             TestWriteMessages(testData);
         }
-
-        public static IDictionary<string, MessageSizeTestData> MessageSizeData => new[]
-        {
-            new MessageSizeTestData("InvocationMessage_WithoutInvocationId", new InvocationMessage("Target", new object[] { 1 }), 15),
-            new MessageSizeTestData("InvocationMessage_WithInvocationId", new InvocationMessage("1", "Target", new object[] { 1 }), 16),
-            new MessageSizeTestData("InvocationMessage_WithInvocationIdAndStreamId", new InvocationMessage("1", "Target", new object[] { 1 }, new string[] { "2" }), 18),
-
-            new MessageSizeTestData("CloseMessage_Empty", CloseMessage.Empty, 5),
-            new MessageSizeTestData("CloseMessage_WithError", new CloseMessage("error"), 10),
-
-            new MessageSizeTestData("StreamItemMessage_WithNullItem", new StreamItemMessage("1", null), 7),
-            new MessageSizeTestData("StreamItemMessage_WithItem", new StreamItemMessage("1", 1), 7),
-
-            new MessageSizeTestData("CompletionMessage_Empty", CompletionMessage.Empty("1"), 7),
-            new MessageSizeTestData("CompletionMessage_WithResult", CompletionMessage.WithResult("1", 1), 8),
-            new MessageSizeTestData("CompletionMessage_WithError", CompletionMessage.WithError("1", "error"), 13),
-
-            new MessageSizeTestData("StreamInvocationMessage", new StreamInvocationMessage("1", "target", Array.Empty<object>()), 15),
-            new MessageSizeTestData("StreamInvocationMessage_WithStreamId", new StreamInvocationMessage("1", "target", Array.Empty<object>(), new [] { "2" }), 17),
-
-            new MessageSizeTestData("CancelInvocationMessage", new CancelInvocationMessage("1"), 6),
-
-            new MessageSizeTestData("PingMessage", PingMessage.Instance, 3),
-        }.ToDictionary(t => t.Name);
-
-        public static IEnumerable<object[]> MessageSizeDataNames => MessageSizeData.Keys.Select(name => new object[] { name });
-
-        [Theory]
-        [MemberData(nameof(MessageSizeDataNames))]
-        public void VerifyMessageSize(string testDataName)
-        {
-            var testData = MessageSizeData[testDataName];
-            Assert.Equal(testData.Size, Write(testData.Message).Length);
-        }
-
-        public class MessageSizeTestData
-        {
-            public string Name { get; }
-            public HubMessage Message { get; }
-            public int Size { get; }
-
-            public MessageSizeTestData(string name, HubMessage message, int size)
-            {
-                Name = name;
-                Message = message;
-                Size = size;
-            }
-
-            public override string ToString() => Name;
-        }
     }
 }