some test simplification to prepare for adding ts client tests for streaming

Author: analogrelay

client-ts/Microsoft.AspNetCore.SignalR.Client.TS.Tests/HubConnection.spec.ts

@@ -3,149 +3,84 @@ import { HubConnection } from "../Microsoft.AspNetCore.SignalR.Client.TS/HubConn
 import { DataReceived, ConnectionClosed } from "../Microsoft.AspNetCore.SignalR.Client.TS/Common"
 import { TransportType, ITransport } from "../Microsoft.AspNetCore.SignalR.Client.TS/Transports"
 
+import { asyncit as it, captureException } from './JasmineUtils';
+
 describe("HubConnection", () => {
-    it("completes pending invocations when stopped", async done => {
-        let connection: IConnection = {
-            start(transportType: TransportType | ITransport): Promise<void> {
-                return Promise.resolve();
-            },
-
-            send(data: any): Promise<void> {
-                return Promise.resolve();
-            },
-
-            stop(): void {
-                if (this.onClosed) {
-                    this.onClosed();
-                }
-            },
-
-            onDataReceived: null,
-            onClosed: null
-        };
+    it("completes pending invocations when stopped", async () => {
+        let connection = new TestConnection();
 
         let hubConnection = new HubConnection(connection);
         var invokePromise = hubConnection.invoke("testMethod");
         hubConnection.stop();
 
-        try {
-            await invokePromise;
-            fail();
-        }
-        catch (e) {
-            expect(e.message).toBe("Invocation cancelled due to connection being closed.");
-        }
-        done();
+        let ex = await captureException(async () => await invokePromise);
+        expect(ex.message).toBe("Invocation cancelled due to connection being closed.");
     });
 
-    it("completes pending invocations when connection is lost", async done => {
-        let connection: IConnection = {
-            start(transportType: TransportType | ITransport): Promise<void> {
-                return Promise.resolve();
-            },
-
-            send(data: any): Promise<void> {
-                return Promise.resolve();
-            },
-
-            stop(): void {
-                if (this.onClosed) {
-                    this.onClosed();
-                }
-            },
-
-            onDataReceived: null,
-            onClosed: null
-        };
+    it("completes pending invocations when connection is lost", async () => {
+        let connection = new TestConnection();
 
         let hubConnection = new HubConnection(connection);
         var invokePromise = hubConnection.invoke("testMethod");
         // Typically this would be called by the transport
         connection.onClosed(new Error("Connection lost"));
 
-        try {
-            await invokePromise;
-            fail();
-        }
-        catch (e) {
-            expect(e.message).toBe("Connection lost");
-        }
-        done();
+        let ex = await captureException(async () => await invokePromise);
+        expect(ex.message).toBe("Connection lost");
     });
 
-    it("sends invocations as nonblocking", async done => {
-        let dataSent: string;
-        let connection: IConnection = {
-            start(transportType: TransportType): Promise<void> {
-                return Promise.resolve();
-            },
-
-            send(data: any): Promise<void> {
-                dataSent = data;
-                return Promise.resolve();
-            },
-
-            stop(): void {
-                if (this.onClosed) {
-                    this.onClosed();
-                }
-            },
-
-            onDataReceived: null,
-            onClosed: null
-        };
+    it("sends invocations as nonblocking", async () => {
+        let connection = new TestConnection();
 
         let hubConnection = new HubConnection(connection);
         let invokePromise = hubConnection.invoke("testMethod");
 
-        expect(JSON.parse(dataSent).nonblocking).toBe(false);
+        expect(connection.sentData.length).toBe(1);
+        expect(JSON.parse(connection.sentData[0]).nonblocking).toBe(false);
 
         // will clean pending promises
         connection.onClosed();
 
-        try {
-            await invokePromise;
-            fail(); // exception is expected because the call has not completed
-        }
-        catch (e) {
-        }
-        done();
+        let ex = await captureException(async () => await invokePromise);
+        // Don't care about the exception
     });
 
-    it("rejects streaming responses made using 'invoke'", async done => {
-        let connection: IConnection = {
-            start(transportType: TransportType): Promise<void> {
-                return Promise.resolve();
-            },
-
-            send(data: any): Promise<void> {
-                return Promise.resolve();
-            },
-
-            stop(): void {
-                if (this.onClosed) {
-                    this.onClosed();
-                }
-            },
-
-            onDataReceived: null,
-            onClosed: null
-        };
+    it("rejects streaming responses made using 'invoke'", async () => {
+        let connection = new TestConnection();
 
         let hubConnection = new HubConnection(connection);
         let invokePromise = hubConnection.invoke("testMethod");
 
         connection.onDataReceived("{ \"type\": 2, \"invocationId\": \"0\", \"result\": null }");
         connection.onClosed();
 
-        try {
-            await invokePromise;
-            fail();
+        let ex = await captureException(async () => await invokePromise);
+        expect(ex.message).toBe("Streaming methods must be invoked using HubConnection.stream");
+    });
+});
+
+class TestConnection implements IConnection {
+    start(transportType: TransportType | ITransport): Promise<void> {
+        return Promise.resolve();
+    };
+
+    send(data: any): Promise<void> {
+        if (this.sentData) {
+            this.sentData.push(data);
         }
-        catch (e) {
-            expect(e.message).toBe("Streaming methods must be invoked using HubConnection.stream");
+        else {
+            this.sentData = [data];
         }
+        return Promise.resolve();
+    };
 
-        done();
-    });
-});
+    stop(): void {
+        if (this.onClosed) {
+            this.onClosed();
+        }
+    };
+
+    onDataReceived: DataReceived;
+    onClosed: ConnectionClosed;
+    sentData: [any];
+};

client-ts/Microsoft.AspNetCore.SignalR.Client.TS.Tests/JasmineUtils.ts

@@ -0,0 +1,21 @@
+export function asyncit(expectation: string, assertion?: () => Promise<any>, timeout?: number): void {
+    let testFunction: (done: DoneFn) => void;
+    if (assertion) {
+        testFunction = done => {
+            assertion()
+                .then(() => done())
+                .catch(() => fail());
+        };
+    }
+
+    it(expectation, testFunction, timeout);
+}
+
+export async function captureException(fn: () => Promise<any>): Promise<Error> {
+    try {
+        await fn();
+        return null;
+    } catch (e) {
+        return e;
+    }
+}

samples/SocketsSample/Hubs/Streaming.cs

@@ -9,13 +9,11 @@ namespace SocketsSample.Hubs
 {
     public class Streaming : Hub
     {
-        [return: Streaming]
         public IObservable<int> ObservableCounter(int count, int delay)
         {
             return new CounterObservable(count, delay);
         }
 
-        [return: Streaming]
         public ReadableChannel<int> ChannelCounter(int count, int delay)
         {
             var channel = Channel.CreateUnbounded<int>();

src/Microsoft.AspNetCore.SignalR/HubEndPoint.cs

@@ -283,10 +283,12 @@ private async Task Invoke(HubMethodDescriptor descriptor, Connection connection,
 
                     if (hasResult && IsStreamed(methodExecutor, result, out var channel))
                     {
+                        _logger.LogTrace("[{connectionId}/{invocationId}] Streaming result of type {resultType}", connection.ConnectionId, invocationMessage.InvocationId, result.GetType().FullName);
                         await StreamResultsAsync(invocationMessage.InvocationId, connection, protocol, channel);
                     }
                     else
                     {
+                        _logger.LogTrace("[{connectionId}/{invocationId}] Sending result of type {resultType}", connection.ConnectionId, invocationMessage.InvocationId, result.GetType().FullName);
                         await SendMessageAsync(connection, protocol, CompletionMessage.WithResult(invocationMessage.InvocationId, result));
                     }
                 }
@@ -343,14 +345,6 @@ private async Task StreamResultsAsync(string invocationId, Connection connection
 
         private bool IsStreamed(ObjectMethodExecutor methodExecutor, object result, out IAsyncEnumerator<object> enumerator)
         {
-            // TODO: Cache attributes the OME?
-            var streamingAttribute = methodExecutor.MethodInfo.ReturnParameter.GetCustomAttribute<StreamingAttribute>();
-            if (streamingAttribute == null)
-            {
-                enumerator = null;
-                return false;
-            }
-
             var observableInterface = result.GetType().GetInterfaces().FirstOrDefault(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IObservable<>));
             if (observableInterface != null)
             {
@@ -364,8 +358,9 @@ private bool IsStreamed(ObjectMethodExecutor methodExecutor, object result, out
             }
             else
             {
-                // This type cannot be streamed
-                throw new NotSupportedException($"Cannot stream results of type: {result.GetType().FullName}");
+                // Not streamed
+                enumerator = null;
+                return false;
             }
         }