Add CancelInvocation support to MsgPack in TS client

src/SignalR/clients/ts/FunctionalTests/TestHub.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Reactive.Linq;
 using System.Text;
+using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Connections;
@@ -21,6 +22,13 @@ public class CustomObject
 
     public class TestHub : Hub
     {
+        private readonly IHubContext<TestHub> _context;
+
+        public TestHub(IHubContext<TestHub> context)
+        {
+            _context = context;
+        }
+
         public string Echo(string message)
         {
             return message;
@@ -51,6 +59,19 @@ public ChannelReader<string> Stream()
             return channel.Reader;
         }
 
+        public ChannelReader<string> InfiniteStream(CancellationToken token)
+        {
+            var channel = Channel.CreateUnbounded<string>();
+            var connectionId = Context.ConnectionId;
+
+            token.Register(async (state) =>
+            {
+                await ((IHubContext<TestHub>)state).Clients.Client(connectionId).SendAsync("StreamCanceled");
+            }, _context);
+
+            return channel.Reader;
+        }
+
         public async Task<string> StreamingConcat(ChannelReader<string> stream)
         {
             var sb = new StringBuilder();

src/SignalR/clients/ts/FunctionalTests/ts/HubConnectionTests.ts

@@ -184,6 +184,39 @@ describe("hubConnection", () => {
                 });
             });
 
+            it("can stream server method and cancel stream", (done) => {
+                const hubConnection = getConnectionBuilder(transportType)
+                    .withHubProtocol(protocol)
+                    .build();
+
+                hubConnection.onclose((error) => {
+                    expect(error).toBe(undefined);
+                    done();
+                });
+
+                hubConnection.on("StreamCanceled", () => {
+                    hubConnection.stop();
+                });
+
+                hubConnection.start().then(() => {
+                    const subscription = hubConnection.stream<string>("InfiniteStream").subscribe({
+                        complete() {
+                        },
+                        error(err) {
+                            fail(err);
+                            hubConnection.stop();
+                        },
+                        next() {
+                        },
+                    });
+
+                    subscription.dispose();
+                }).catch((e) => {
+                    fail(e);
+                    done();
+                });
+            });
+
             it("rethrows an exception from the server when invoking", (done) => {
                 const errorMessage = "An unexpected error occurred invoking 'ThrowException' on the server. InvalidOperationException: An error occurred.";
                 const hubConnection = getConnectionBuilder(transportType)

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

@@ -4,7 +4,8 @@
 import { Buffer } from "buffer";
 import * as msgpack5 from "msgpack5";
 
-import { CompletionMessage, HubMessage, IHubProtocol, ILogger, InvocationMessage, LogLevel, MessageHeaders, MessageType, NullLogger, StreamInvocationMessage, StreamItemMessage, TransferFormat } from "@aspnet/signalr";
+import { CancelInvocationMessage, CompletionMessage, HubMessage, IHubProtocol, ILogger, InvocationMessage,
+    LogLevel, MessageHeaders, MessageType, NullLogger, StreamInvocationMessage, StreamItemMessage, TransferFormat } from "@aspnet/signalr";
 
 import { BinaryMessageFormat } from "./BinaryMessageFormat";
 import { isArrayBuffer } from "./Utils";
@@ -75,6 +76,8 @@ export class MessagePackHubProtocol implements IHubProtocol {
                 return this.writeCompletion(message as CompletionMessage);
             case MessageType.Ping:
                 return BinaryMessageFormat.write(SERIALIZED_PING_MESSAGE);
+            case MessageType.CancelInvocation:
+                return this.writeCancelInvocation(message as CancelInvocationMessage);
             default:
                 throw new Error("Invalid message type.");
         }
@@ -257,6 +260,13 @@ export class MessagePackHubProtocol implements IHubProtocol {
         return BinaryMessageFormat.write(payload.slice());
     }
 
+    private writeCancelInvocation(cancelInvocationMessage: CancelInvocationMessage): ArrayBuffer {
+        const msgpack = msgpack5();
+        const payload = msgpack.encode([MessageType.CancelInvocation, cancelInvocationMessage.headers || {}, cancelInvocationMessage.invocationId]);
+
+        return BinaryMessageFormat.write(payload.slice());
+    }
+
     private readHeaders(properties: any): MessageHeaders {
         const headers: MessageHeaders = properties[1] as MessageHeaders;
         if (typeof headers !== "object") {

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

@@ -202,4 +202,19 @@ describe("MessagePackHubProtocol", () => {
         const buffer = new MessagePackHubProtocol().writeMessage({ type: MessageType.Ping });
         expect(new Uint8Array(buffer)).toEqual(payload);
     });
+
+    it("can write cancel message", () => {
+        const payload = new Uint8Array([
+            0x07, // length prefix
+            0x93, // message array length = 1 (fixarray)
+            0x05, // type = 5 = CancelInvocation (fixnum)
+            0x80, // headers
+            0xa3, // invocationID = string length 3
+            0x61, // a
+            0x62, // b
+            0x63, // c
+        ]);
+        const buffer = new MessagePackHubProtocol().writeMessage({ type: MessageType.CancelInvocation, invocationId: "abc" });
+        expect(new Uint8Array(buffer)).toEqual(payload);
+    });
 });

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

@@ -156,13 +156,16 @@ export class HubConnection {
         const [streams, streamIds] = this.replaceStreamingParams(args);
         const invocationDescriptor = this.createStreamInvocation(methodName, args, streamIds);
 
+        let promiseQueue: Promise<void>;
         const subject = new Subject<T>();
         subject.cancelCallback = () => {
             const cancelInvocation: CancelInvocationMessage = this.createCancelInvocation(invocationDescriptor.invocationId);
 
             delete this.callbacks[invocationDescriptor.invocationId];
 
-            return this.sendWithProtocol(cancelInvocation);
+            return promiseQueue.then(() => {
+                return this.sendWithProtocol(cancelInvocation);
+            });
         };
 
         this.callbacks[invocationDescriptor.invocationId] = (invocationEvent: CompletionMessage | StreamItemMessage | null, error?: Error) => {
@@ -183,7 +186,7 @@ export class HubConnection {
             }
         };
 
-        const promiseQueue = this.sendWithProtocol(invocationDescriptor)
+        promiseQueue = this.sendWithProtocol(invocationDescriptor)
             .catch((e) => {
                 subject.error(e);
                 delete this.callbacks[invocationDescriptor.invocationId];

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

@@ -1075,6 +1075,8 @@ describe("HubConnection", () => {
                     // Observer should no longer receive messages
                     expect(observer.itemsReceived).toEqual([1]);
 
+                    // Close message sent asynchronously so we need to wait
+                    await delay(50);
                     // Verify the cancel is sent (+ handshake)
                     expect(connection.sentData.length).toBe(3);
                     expect(JSON.parse(connection.sentData[2])).toEqual({