[TS Client] Test message size (#15099)

Author: BrennanConroy

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,
-        invocationMessage.target, invocationMessage.arguments, invocationMessage.streamIds]);
+        let payload: any;
+        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,
-        streamInvocationMessage.target, streamInvocationMessage.arguments, streamInvocationMessage.streamIds]);
+        let payload: any;
+        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 {
-                arguments: args,
-                streamIds,
-                target: methodName,
-                type: MessageType.Invocation,
-            };
+            if (streamIds.length !== 0) {
+                return {
+                    arguments: args,
+                    streamIds,
+                    target: methodName,
+                    type: MessageType.Invocation,
+                };
+            } else {
+                return {
+                    arguments: args,
+                    target: methodName,
+                    type: MessageType.Invocation,
+                };
+            }
         } else {
             const invocationId = this.invocationId;
             this.invocationId++;
 
-            return {
-                arguments: args,
-                invocationId: invocationId.toString(),
-                streamIds,
-                target: methodName,
-                type: MessageType.Invocation,
-            };
+            if (streamIds.length !== 0) {
+                return {
+                    arguments: args,
+                    invocationId: invocationId.toString(),
+                    streamIds,
+                    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 {
-            arguments: args,
-            invocationId: invocationId.toString(),
-            streamIds,
-            target: methodName,
-            type: MessageType.StreamInvocation,
-        };
+        if (streamIds.length !== 0) {
+            return {
+                arguments: args,
+                invocationId: invocationId.toString(),
+                streamIds,
+                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 @@ public void ReadCaseInsensitivePropertiesByDefault()
 
         [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];