Merge branch 'release/2.1' into jamesnk/test-server-logging

clients/ts/FunctionalTests/rollup.config.js

@@ -8,7 +8,7 @@ import commonjs from 'rollup-plugin-commonjs'
 import resolve from 'rollup-plugin-node-resolve'
 
 export default {
-    input: path.join(__dirname, "obj", "js", "index.js"),
+    input: path.join(__dirname, "obj", "js", "FunctionalTests", "ts", "index.js"),
     output: {
         file: path.join(__dirname, "wwwroot", "dist", "signalr-functional-tests.js"),
         format: "iife",

clients/ts/FunctionalTests/ts/HubConnectionTests.ts

@@ -1,7 +1,7 @@
 // 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 { DefaultHttpClient, HttpClient, HttpRequest, HttpResponse, HttpTransportType, HubConnection, IHubConnectionOptions, JsonHubProtocol, LogLevel } from "@aspnet/signalr";
+import { DefaultHttpClient, HttpClient, HttpRequest, HttpResponse, HttpTransportType, HubConnection, IHubConnectionOptions, IStreamSubscriber, JsonHubProtocol, LogLevel } from "@aspnet/signalr";
 import { MessagePackHubProtocol } from "@aspnet/signalr-protocol-msgpack";
 
 import { eachTransport, eachTransportAndProtocol } from "./Common";
@@ -114,15 +114,15 @@ describe("hubConnection", () => {
                 const received = [];
                 hubConnection.start().then(() => {
                     hubConnection.stream("Stream").subscribe({
-                        complete: function complete() {
+                        complete() {
                             expect(received).toEqual(["a", "b", "c"]);
                             hubConnection.stop();
                         },
-                        error: function error(err) {
+                        error(err) {
                             fail(err);
                             hubConnection.stop();
                         },
-                        next: function next(item) {
+                        next(item) {
                             received.push(item);
                         },
                     });
@@ -215,16 +215,16 @@ describe("hubConnection", () => {
 
                 hubConnection.start().then(() => {
                     hubConnection.stream("StreamThrowException", "An error occurred.").subscribe({
-                        complete: function complete() {
+                        complete() {
                             hubConnection.stop();
                             fail();
                         },
-                        error: function error(err) {
+                        error(err) {
                             expect(err.message).toEqual(errorMessage);
                             hubConnection.stop();
                             done();
                         },
-                        next: function next(item) {
+                        next(item) {
                             hubConnection.stop();
                             fail();
                         },
@@ -244,16 +244,16 @@ describe("hubConnection", () => {
 
                 hubConnection.start().then(() => {
                     hubConnection.stream("Echo", "42").subscribe({
-                        complete: function complete() {
+                        complete() {
                             hubConnection.stop();
                             fail();
                         },
-                        error: function error(err) {
+                        error(err) {
                             expect(err.message).toEqual("The client attempted to invoke the non-streaming 'Echo' method with a streaming invocation.");
                             hubConnection.stop();
                             done();
                         },
-                        next: function next(item) {
+                        next(item) {
                             hubConnection.stop();
                             fail();
                         },

clients/ts/FunctionalTests/ts/TestLogger.ts

@@ -1,4 +1,8 @@
-import { ConsoleLogger, ILogger, LogLevel } from "@aspnet/signalr";
+import { ILogger, LogLevel } from "@aspnet/signalr";
+
+// Since JavaScript modules are file-based, we can just pull in utilities from the
+// main library directly even if they aren't exported.
+import { ConsoleLogger } from "../../signalr/src/Utils";
 
 export class TestLog {
     public messages: Array<[Date, LogLevel, string]> = [];

clients/ts/signalr-protocol-msgpack/spec/MessagePackHubProtocol.spec.ts

@@ -14,7 +14,7 @@ describe("MessageHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new MessagePackHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -27,7 +27,7 @@ describe("MessageHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new MessagePackHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -42,7 +42,7 @@ describe("MessageHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new MessagePackHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -56,7 +56,7 @@ describe("MessageHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new MessagePackHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -104,7 +104,7 @@ describe("MessageHubProtocol", () => {
         } as CompletionMessage],
     ] as Array<[number[], CompletionMessage]>).forEach(([payload, expectedMessage]) =>
         it("can read Completion message", () => {
-            const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, new NullLogger());
+            const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, NullLogger.instance);
             expect(messages).toEqual([expectedMessage]);
         }));
 
@@ -125,7 +125,7 @@ describe("MessageHubProtocol", () => {
         } as StreamItemMessage],
     ] as Array<[number[], StreamItemMessage]>).forEach(([payload, expectedMessage]) =>
         it("can read StreamItem message", () => {
-            const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, new NullLogger());
+            const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, NullLogger.instance);
             expect(messages).toEqual([expectedMessage]);
         }));
 
@@ -141,7 +141,7 @@ describe("MessageHubProtocol", () => {
         } as StreamItemMessage],
     ] as Array<[number[], StreamItemMessage]>).forEach(([payload, expectedMessage]) =>
         it("can read message with headers", () => {
-            const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, new NullLogger());
+            const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, NullLogger.instance);
             expect(messages).toEqual([expectedMessage]);
         }));
 
@@ -157,7 +157,7 @@ describe("MessageHubProtocol", () => {
         ["Completion message with missing error", [0x05, 0x94, 0x03, 0x80, 0xa0, 0x03], new Error("Invalid payload for Completion message.")],
     ] as Array<[string, number[], Error]>).forEach(([name, payload, expectedError]) =>
         it("throws for " + name, () => {
-            expect(() => new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, new NullLogger()))
+            expect(() => new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, NullLogger.instance))
                 .toThrow(expectedError);
         }));
 
@@ -165,7 +165,7 @@ describe("MessageHubProtocol", () => {
         const payload = [
             0x08, 0x94, 0x02, 0x80, 0xa3, 0x61, 0x62, 0x63, 0x08,
             0x0b, 0x95, 0x03, 0x80, 0xa3, 0x61, 0x62, 0x63, 0x03, 0xa2, 0x4f, 0x4b];
-        const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, new NullLogger());
+        const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, NullLogger.instance);
         expect(messages).toEqual([
             {
                 headers: {},
@@ -189,7 +189,7 @@ describe("MessageHubProtocol", () => {
             0x91, // message array length = 1 (fixarray)
             0x06, // type = 6 = Ping (fixnum)
         ];
-        const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, new NullLogger());
+        const messages = new MessagePackHubProtocol().parseMessages(new Uint8Array(payload).buffer, NullLogger.instance);
         expect(messages).toEqual([
             {
                 type: MessageType.Ping,

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

@@ -15,7 +15,7 @@ export class MessagePackHubProtocol implements IHubProtocol {
 
     public parseMessages(input: ArrayBuffer, logger: ILogger): HubMessage[] {
         if (logger === null) {
-            logger = new NullLogger();
+            logger = NullLogger.instance;
         }
         return BinaryMessageFormat.parse(input).map((m) => this.parseMessage(m, logger));
     }

clients/ts/signalr/spec/Common.ts

@@ -1,7 +1,7 @@
 // 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 { ITransport, HttpTransportType } from "../src/ITransport";
+import { HttpTransportType, ITransport } from "../src/ITransport";
 
 export function eachTransport(action: (transport: HttpTransportType) => void) {
     const transportTypes = [

clients/ts/signalr/spec/HttpConnection.spec.ts

@@ -1,11 +1,10 @@
 // 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 { DataReceived, TransportClosed } from "../src/Common";
 import { HttpConnection } from "../src/HttpConnection";
 import { IHttpConnectionOptions } from "../src/HttpConnection";
 import { HttpResponse } from "../src/index";
-import { ITransport, TransferFormat, HttpTransportType } from "../src/ITransport";
+import { HttpTransportType, ITransport, TransferFormat } from "../src/ITransport";
 import { eachEndpointUrl, eachTransport } from "./Common";
 import { TestHttpClient } from "./TestHttpClient";
 
@@ -423,13 +422,6 @@ describe("HttpConnection", () => {
     });
 
     describe("startAsync", () => {
-        it("throws if no TransferFormat is provided", async () => {
-            // Force TypeScript to let us call start incorrectly
-            const connection: any = new HttpConnection("http://tempuri.org", commonOptions);
-
-            expect(() => connection.start()).toThrowError("The 'transferFormat' argument is required.");
-        });
-
         it("throws if an unsupported TransferFormat is provided", async () => {
             // Force TypeScript to let us call start incorrectly
             const connection: any = new HttpConnection("http://tempuri.org", commonOptions);

clients/ts/signalr/spec/HubConnection.spec.ts

@@ -1,16 +1,15 @@
 // 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 { ConnectionClosed, DataReceived } from "../src/Common";
 import { HubConnection } from "../src/HubConnection";
+import { IHubConnectionOptions } from "../src/HubConnection";
 import { IConnection } from "../src/IConnection";
 import { HubMessage, IHubProtocol, MessageType } from "../src/IHubProtocol";
 import { ILogger, LogLevel } from "../src/ILogger";
-import { Observer } from "../src/Observable";
+import { HttpTransportType, ITransport, TransferFormat } from "../src/ITransport";
+import { IStreamSubscriber } from "../src/Stream";
 import { TextMessageFormat } from "../src/TextMessageFormat";
-import { ITransport, TransferFormat, HttpTransportType } from "../src/ITransport";
 
-import { IHubConnectionOptions } from "../src/HubConnection";
 import { asyncit as it, captureException, delay, PromiseSource } from "./Utils";
 
 const commonOptions: IHubConnectionOptions = {
@@ -124,7 +123,7 @@ describe("HubConnection", () => {
             let receivedProcotolData: ArrayBuffer;
 
             const mockProtocol = new TestProtocol(TransferFormat.Binary);
-            mockProtocol.onreceive = (d) => receivedProcotolData = d;
+            mockProtocol.onreceive = (d) => receivedProcotolData = d as ArrayBuffer;
 
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, { logger: null, protocol: mockProtocol });
@@ -136,7 +135,7 @@ describe("HubConnection", () => {
                 0x69, 0x6e, 0x76, 0x6f, 0x6b, 0x65, 0x20, 0x74, 0x68, 0x65, 0x20, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69,
                 0x6e, 0x67, 0x20, 0x27, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x27, 0x20, 0x6d,
                 0x65, 0x74, 0x68, 0x6f, 0x64, 0x20, 0x69, 0x6e, 0x20, 0x61, 0x20, 0x6e, 0x6f, 0x6e, 0x2d, 0x73, 0x74, 0x72,
-                0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x20, 0x66, 0x61, 0x73, 0x68, 0x69, 0x6f, 0x6e, 0x2e
+                0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x20, 0x66, 0x61, 0x73, 0x68, 0x69, 0x6f, 0x6e, 0x2e,
             ];
 
             connection.receiveBinary(new Uint8Array(data).buffer);
@@ -149,7 +148,7 @@ describe("HubConnection", () => {
             let receivedProcotolData: string;
 
             const mockProtocol = new TestProtocol(TransferFormat.Text);
-            mockProtocol.onreceive = (d) => receivedProcotolData = d;
+            mockProtocol.onreceive = (d) => receivedProcotolData = d as string;
 
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, { logger: null, protocol: mockProtocol });
@@ -256,8 +255,8 @@ describe("HubConnection", () => {
             connection.receiveHandshakeResponse();
 
             const handler = () => { };
-            hubConnection.on('message', handler);
+            hubConnection.on("message", handler);
-            hubConnection.off('message', handler);
+            hubConnection.off("message", handler);
 
             connection.receive({
                 arguments: ["test"],
@@ -648,9 +647,7 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
 
             const hubConnection = new HubConnection(connection, commonOptions);
-            const observer = hubConnection.stream("testMethod").subscribe({
+            const observer = hubConnection.stream("testMethod").subscribe(NullSubscriber.instance);
-                next: (val) => { },
-            });
 
             // Typically this would be called by the transport
             // triggers observer.error()
@@ -661,9 +658,7 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
 
             const hubConnection = new HubConnection(connection, commonOptions);
-            const observer = hubConnection.stream("testMethod").subscribe({
+            const observer = hubConnection.stream("testMethod").subscribe(NullSubscriber.instance);
-                next: (val) => { },
-            });
 
             // Send completion to trigger observer.complete()
             // Expectation is connection.receive will not to throw
@@ -843,7 +838,7 @@ class TestConnection implements IConnection {
     }
 
     public receiveHandshakeResponse(error?: string): void {
-        this.receive({error: error});
+        this.receive({ error });
     }
 
     public receive(data: any): void {
@@ -859,8 +854,8 @@ class TestConnection implements IConnection {
         this.onreceive(data);
     }
 
-    public onreceive: DataReceived;
+    public onreceive: (data: string | ArrayBuffer) => void;
-    public onclose: ConnectionClosed;
+    public onclose: (error?: Error) => void;
     public sentData: any[];
     public lastInvocationId: string;
 }
@@ -871,7 +866,7 @@ class TestProtocol implements IHubProtocol {
 
     public readonly transferFormat: TransferFormat;
 
-    public onreceive: DataReceived;
+    public onreceive: (data: string | ArrayBuffer) => void;
 
     constructor(transferFormat: TransferFormat) {
         this.transferFormat = transferFormat;
@@ -890,7 +885,8 @@ class TestProtocol implements IHubProtocol {
     }
 }
 
-class TestObserver implements Observer<any> {
+class TestObserver implements IStreamSubscriber<any> {
+    public readonly closed: boolean;
     public itemsReceived: [any];
     private itemsSource: PromiseSource<[any]>;
 
@@ -915,3 +911,17 @@ class TestObserver implements Observer<any> {
         this.itemsSource.resolve(this.itemsReceived);
     }
 }
+
+class NullSubscriber<T> implements IStreamSubscriber<T> {
+    public static instance: NullSubscriber<any> = new NullSubscriber();
+
+    private constructor() {
+    }
+
+    public next(value: T): void {
+    }
+    public error(err: any): void {
+    }
+    public complete(): void {
+    }
+}

clients/ts/signalr/spec/JsonHubProtocol.spec.ts

@@ -16,7 +16,7 @@ describe("JsonHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new JsonHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -29,7 +29,7 @@ describe("JsonHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new JsonHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -44,7 +44,7 @@ describe("JsonHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new JsonHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -58,7 +58,7 @@ describe("JsonHubProtocol", () => {
         } as InvocationMessage;
 
         const protocol = new JsonHubProtocol();
-        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), new NullLogger());
+        const parsedMessages = protocol.parseMessages(protocol.writeMessage(invocation), NullLogger.instance);
         expect(parsedMessages).toEqual([invocation]);
     });
 
@@ -106,7 +106,7 @@ describe("JsonHubProtocol", () => {
         } as CompletionMessage],
     ] as Array<[string, CompletionMessage]>).forEach(([payload, expectedMessage]) =>
         it("can read Completion message", () => {
-            const messages = new JsonHubProtocol().parseMessages(payload, new NullLogger());
+            const messages = new JsonHubProtocol().parseMessages(payload, NullLogger.instance);
             expect(messages).toEqual([expectedMessage]);
         }));
 
@@ -127,7 +127,7 @@ describe("JsonHubProtocol", () => {
         } as StreamItemMessage],
     ] as Array<[string, StreamItemMessage]>).forEach(([payload, expectedMessage]) =>
         it("can read StreamItem message", () => {
-            const messages = new JsonHubProtocol().parseMessages(payload, new NullLogger());
+            const messages = new JsonHubProtocol().parseMessages(payload, NullLogger.instance);
             expect(messages).toEqual([expectedMessage]);
         }));
 
@@ -143,7 +143,7 @@ describe("JsonHubProtocol", () => {
         } as StreamItemMessage],
     ] as Array<[string, StreamItemMessage]>).forEach(([payload, expectedMessage]) =>
         it("can read message with headers", () => {
-            const messages = new JsonHubProtocol().parseMessages(payload, new NullLogger());
+            const messages = new JsonHubProtocol().parseMessages(payload, NullLogger.instance);
             expect(messages).toEqual([expectedMessage]);
         }));
 
@@ -160,13 +160,13 @@ describe("JsonHubProtocol", () => {
         ["Completion message with non-string error", `{"type":3,"invocationId":"1","error":21}${TextMessageFormat.RecordSeparator}`, new Error("Invalid payload for Completion message.")],
     ] as Array<[string, string, Error]>).forEach(([name, payload, expectedError]) =>
         it("throws for " + name, () => {
-            expect(() => new JsonHubProtocol().parseMessages(payload, new NullLogger()))
+            expect(() => new JsonHubProtocol().parseMessages(payload, NullLogger.instance))
                 .toThrow(expectedError);
         }));
 
     it("can read multiple messages", () => {
         const payload = `{"type":2, "invocationId": "abc", "headers": {}, "item": 8}${TextMessageFormat.RecordSeparator}{"type":3, "invocationId": "abc", "headers": {}, "result": "OK", "error": null}${TextMessageFormat.RecordSeparator}`;
-        const messages = new JsonHubProtocol().parseMessages(payload, new NullLogger());
+        const messages = new JsonHubProtocol().parseMessages(payload, NullLogger.instance);
         expect(messages).toEqual([
             {
                 headers: {},
@@ -186,7 +186,7 @@ describe("JsonHubProtocol", () => {
 
     it("can read ping message", () => {
         const payload = `{"type":6}${TextMessageFormat.RecordSeparator}`;
-        const messages = new JsonHubProtocol().parseMessages(payload, new NullLogger());
+        const messages = new JsonHubProtocol().parseMessages(payload, NullLogger.instance);
         expect(messages).toEqual([
             {
                 type: MessageType.Ping,

clients/ts/signalr/spec/LoggerFactory.spec.ts

@@ -1,24 +0,0 @@
-// 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 { ILogger, LogLevel } from "../src/ILogger";
-import { LoggerFactory } from "../src/Loggers";
-
-describe("LoggerFactory", () => {
-    it("creates ConsoleLogger when no logging specified", () => {
-        expect(LoggerFactory.createLogger().constructor.name).toBe("ConsoleLogger");
-    });
-
-    it("creates NullLogger when logging is set to null", () => {
-        expect(LoggerFactory.createLogger(null).constructor.name).toBe("NullLogger");
-    });
-
-    it("creates ConsoleLogger when log level specified", () => {
-        expect(LoggerFactory.createLogger(LogLevel.Information).constructor.name).toBe("ConsoleLogger");
-    });
-
-    it("does not create its own logger if the user provides one", () => {
-        const customLogger: ILogger = { log: (logLevel) => {} };
-        expect(LoggerFactory.createLogger(customLogger)).toBe(customLogger);
-    });
-});

clients/ts/signalr/src/Common.ts

@@ -1,6 +0,0 @@
-// 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.
-
-export declare type DataReceived = (data: any) => void;
-export declare type ConnectionClosed = (e?: Error) => void;
-export declare type TransportClosed = (e?: Error) => void;

clients/ts/signalr/src/HttpConnection.ts

@@ -1,15 +1,13 @@
 // 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 { ConnectionClosed, DataReceived } from "./Common";
 import { DefaultHttpClient, HttpClient } from "./HttpClient";
 import { IConnection } from "./IConnection";
 import { ILogger, LogLevel } from "./ILogger";
 import { HttpTransportType, ITransport, TransferFormat } from "./ITransport";
-import { LoggerFactory } from "./Loggers";
 import { LongPollingTransport } from "./LongPollingTransport";
 import { ServerSentEventsTransport } from "./ServerSentEventsTransport";
-import { Arg } from "./Utils";
+import { Arg, createLogger } from "./Utils";
 import { WebSocketTransport } from "./WebSocketTransport";
 
 export interface IHttpConnectionOptions {
@@ -49,11 +47,13 @@ export class HttpConnection implements IConnection {
     private stopError?: Error;
 
     public readonly features: any = {};
+    public onreceive: (data: string | ArrayBuffer) => void;
+    public onclose: (e?: Error) => void;
 
     constructor(url: string, options: IHttpConnectionOptions = {}) {
         Arg.isRequired(url, "url");
 
-        this.logger = LoggerFactory.createLogger(options.logger);
+        this.logger = createLogger(options.logger);
         this.baseUrl = this.resolveUrl(url);
 
         options = options || {};
@@ -65,11 +65,14 @@ export class HttpConnection implements IConnection {
         this.options = options;
     }
 
-    public start(transferFormat: TransferFormat): Promise<void> {
+    public start(): Promise<void>;
-        Arg.isRequired(transferFormat, "transferFormat");
+    public start(transferFormat: TransferFormat): Promise<void>;
+    public start(transferFormat?: TransferFormat): Promise<void> {
+        transferFormat = transferFormat || TransferFormat.Binary;
+
         Arg.isIn(transferFormat, TransferFormat, "transferFormat");
 
-        this.logger.log(LogLevel.Trace, `Starting connection with transfer format '${TransferFormat[transferFormat]}'.`);
+        this.logger.log(LogLevel.Debug, `Starting connection with transfer format '${TransferFormat[transferFormat]}'.`);
 
         if (this.connectionState !== ConnectionState.Disconnected) {
             return Promise.reject(new Error("Cannot start a connection that is not in the 'Disconnected' state."));
@@ -81,6 +84,31 @@ export class HttpConnection implements IConnection {
         return this.startPromise;
     }
 
+    public send(data: string | ArrayBuffer): Promise<void> {
+        if (this.connectionState !== ConnectionState.Connected) {
+            throw new Error("Cannot send data if the connection is not in the 'Connected' State.");
+        }
+
+        return this.transport.send(data);
+    }
+
+    public async stop(error?: Error): Promise<void> {
+        this.connectionState = ConnectionState.Disconnected;
+
+        try {
+            await this.startPromise;
+        } catch (e) {
+            // this exception is returned to the user as a rejected Promise from the start method
+        }
+
+        // The transport's onclose will trigger stopConnection which will run our onclose event.
+        if (this.transport) {
+            this.stopError = error;
+            await this.transport.stop();
+            this.transport = null;
+        }
+    }
+
     private async startInternal(transferFormat: TransferFormat): Promise<void> {
         try {
             if (this.options.transport === HttpTransportType.WebSockets) {
@@ -127,7 +155,7 @@ export class HttpConnection implements IConnection {
 
     private async getNegotiationResponse(headers: any): Promise<INegotiateResponse> {
         const negotiateUrl = this.resolveNegotiateUrl(this.baseUrl);
-        this.logger.log(LogLevel.Trace, `Sending negotiation request: ${negotiateUrl}`);
+        this.logger.log(LogLevel.Debug, `Sending negotiation request: ${negotiateUrl}`);
         try {
             const response = await this.httpClient.post(negotiateUrl, {
                 content: "",
@@ -148,7 +176,7 @@ export class HttpConnection implements IConnection {
     private async createTransport(requestedTransport: HttpTransportType | ITransport, negotiateResponse: INegotiateResponse, requestedTransferFormat: TransferFormat, headers: any): Promise<void> {
         this.updateConnectionId(negotiateResponse);
         if (this.isITransport(requestedTransport)) {
-            this.logger.log(LogLevel.Trace, "Connection was provided an instance of ITransport, using that directly.");
+            this.logger.log(LogLevel.Debug, "Connection was provided an instance of ITransport, using that directly.");
             this.transport = requestedTransport;
             await this.transport.connect(this.url, requestedTransferFormat);
 
@@ -199,23 +227,23 @@ export class HttpConnection implements IConnection {
     private resolveTransport(endpoint: IAvailableTransport, requestedTransport: HttpTransportType, requestedTransferFormat: TransferFormat): HttpTransportType | null {
         const transport = HttpTransportType[endpoint.transport];
         if (transport === null || transport === undefined) {
-            this.logger.log(LogLevel.Trace, `Skipping transport '${endpoint.transport}' because it is not supported by this client.`);
+            this.logger.log(LogLevel.Debug, `Skipping transport '${endpoint.transport}' because it is not supported by this client.`);
         } else {
             const transferFormats = endpoint.transferFormats.map((s) => TransferFormat[s]);
             if (!requestedTransport || transport === requestedTransport) {
                 if (transferFormats.indexOf(requestedTransferFormat) >= 0) {
                     if ((transport === HttpTransportType.WebSockets && typeof WebSocket === "undefined") ||
                         (transport === HttpTransportType.ServerSentEvents && typeof EventSource === "undefined")) {
-                        this.logger.log(LogLevel.Trace, `Skipping transport '${HttpTransportType[transport]}' because it is not supported in your environment.'`);
+                        this.logger.log(LogLevel.Debug, `Skipping transport '${HttpTransportType[transport]}' because it is not supported in your environment.'`);
                     } else {
-                        this.logger.log(LogLevel.Trace, `Selecting transport '${HttpTransportType[transport]}'`);
+                        this.logger.log(LogLevel.Debug, `Selecting transport '${HttpTransportType[transport]}'`);
                         return transport;
                     }
                 } else {
-                    this.logger.log(LogLevel.Trace, `Skipping transport '${HttpTransportType[transport]}' because it does not support the requested transfer format '${TransferFormat[requestedTransferFormat]}'.`);
+                    this.logger.log(LogLevel.Debug, `Skipping transport '${HttpTransportType[transport]}' because it does not support the requested transfer format '${TransferFormat[requestedTransferFormat]}'.`);
                 }
             } else {
-                this.logger.log(LogLevel.Trace, `Skipping transport '${HttpTransportType[transport]}' because it was disabled by the client.`);
+                this.logger.log(LogLevel.Debug, `Skipping transport '${HttpTransportType[transport]}' because it was disabled by the client.`);
             }
         }
         return null;
@@ -233,31 +261,6 @@ export class HttpConnection implements IConnection {
         return false;
     }
 
-    public send(data: any): Promise<void> {
-        if (this.connectionState !== ConnectionState.Connected) {
-            throw new Error("Cannot send data if the connection is not in the 'Connected' State.");
-        }
-
-        return this.transport.send(data);
-    }
-
-    public async stop(error?: Error): Promise<void> {
-        this.connectionState = ConnectionState.Disconnected;
-
-        try {
-            await this.startPromise;
-        } catch (e) {
-            // this exception is returned to the user as a rejected Promise from the start method
-        }
-
-        // The transport's onclose will trigger stopConnection which will run our onclose event.
-        if (this.transport) {
-            this.stopError = error;
-            await this.transport.stop();
-            this.transport = null;
-        }
-    }
-
     private async stopConnection(error?: Error): Promise<void> {
         this.transport = null;
 
@@ -309,7 +312,4 @@ export class HttpConnection implements IConnection {
         negotiateUrl += index === -1 ? "" : url.substring(index);
         return negotiateUrl;
     }
-
-    public onreceive: DataReceived;
-    public onclose: ConnectionClosed;
 }

clients/ts/signalr/src/HubConnection.ts

@@ -1,16 +1,16 @@
 // 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 { ConnectionClosed } from "./Common";
 import { HandshakeProtocol, HandshakeRequestMessage, HandshakeResponseMessage } from "./HandshakeProtocol";
 import { HttpConnection, IHttpConnectionOptions } from "./HttpConnection";
 import { IConnection } from "./IConnection";
 import { CancelInvocationMessage, CompletionMessage, HubMessage, IHubProtocol, InvocationMessage, MessageType, StreamInvocationMessage, StreamItemMessage } from "./IHubProtocol";
 import { ILogger, LogLevel } from "./ILogger";
 import { JsonHubProtocol } from "./JsonHubProtocol";
-import { ConsoleLogger, LoggerFactory, NullLogger } from "./Loggers";
+import { NullLogger } from "./Loggers";
-import { Observable, Subject } from "./Observable";
+import { IStreamResult } from "./Stream";
 import { TextMessageFormat } from "./TextMessageFormat";
+import { createLogger, Subject } from "./Utils";
 
 export { JsonHubProtocol };
 
@@ -29,7 +29,7 @@ export class HubConnection {
     private callbacks: { [invocationId: string]: (invocationEvent: StreamItemMessage | CompletionMessage, error?: Error) => void };
     private methods: { [name: string]: Array<(...args: any[]) => void> };
     private id: number;
-    private closedCallbacks: ConnectionClosed[];
+    private closedCallbacks: Array<(error?: Error) => void>;
     private timeoutHandle: NodeJS.Timer;
     private timeoutInMilliseconds: number;
     private receivedHandshakeResponse: boolean;
@@ -50,7 +50,7 @@ export class HubConnection {
             this.connection = urlOrConnection;
         }
 
-        this.logger = LoggerFactory.createLogger(options.logger);
+        this.logger = createLogger(options.logger);
 
         this.connection.onreceive = (data: any) => this.processIncomingData(data);
         this.connection.onclose = (error?: Error) => this.connectionClosed(error);
@@ -61,132 +61,19 @@ export class HubConnection {
         this.id = 0;
     }
 
-    private processIncomingData(data: any) {
-        this.cleanupTimeout();
-
-        if (!this.receivedHandshakeResponse) {
-            data = this.processHandshakeResponse(data);
-            this.receivedHandshakeResponse = true;
-        }
-
-        // Data may have all been read when processing handshake response
-        if (data) {
-            // Parse the messages
-            const messages = this.protocol.parseMessages(data, this.logger);
-
-            for (const message of messages) {
-                switch (message.type) {
-                    case MessageType.Invocation:
-                        this.invokeClientMethod(message);
-                        break;
-                    case MessageType.StreamItem:
-                    case MessageType.Completion:
-                        const callback = this.callbacks[message.invocationId];
-                        if (callback != null) {
-                            if (message.type === MessageType.Completion) {
-                                delete this.callbacks[message.invocationId];
-                            }
-                            callback(message);
-                        }
-                        break;
-                    case MessageType.Ping:
-                        // Don't care about pings
-                        break;
-                    case MessageType.Close:
-                        this.logger.log(LogLevel.Information, "Close message received from server.");
-                        this.connection.stop(message.error ? new Error("Server returned an error on close: " + message.error) : null);
-                        break;
-                    default:
-                        this.logger.log(LogLevel.Warning, "Invalid message type: " + message.type);
-                        break;
-                }
-            }
-        }
-
-        this.configureTimeout();
-    }
-
-    private processHandshakeResponse(data: any): any {
-        let responseMessage: HandshakeResponseMessage;
-        let remainingData: any;
-
-        try {
-            [remainingData, responseMessage] = this.handshakeProtocol.parseHandshakeResponse(data);
-        } catch (e) {
-            const message = "Error parsing handshake response: " + e;
-            this.logger.log(LogLevel.Error, message);
-
-            const error = new Error(message);
-            this.connection.stop(error);
-            throw error;
-        }
-        if (responseMessage.error) {
-            const message = "Server returned handshake error: " + responseMessage.error;
-            this.logger.log(LogLevel.Error, message);
-            this.connection.stop(new Error(message));
-        } else {
-            this.logger.log(LogLevel.Trace, "Server handshake complete.");
-        }
-
-        return remainingData;
-    }
-
-    private configureTimeout() {
-        if (!this.connection.features || !this.connection.features.inherentKeepAlive) {
-            // Set the timeout timer
-            this.timeoutHandle = setTimeout(() => this.serverTimeout(), this.timeoutInMilliseconds);
-        }
-    }
-
-    private serverTimeout() {
-        // The server hasn't talked to us in a while. It doesn't like us anymore ... :(
-        // Terminate the connection
-        this.connection.stop(new Error("Server timeout elapsed without receiving a message from the server."));
-    }
-
-    private invokeClientMethod(invocationMessage: InvocationMessage) {
-        const methods = this.methods[invocationMessage.target.toLowerCase()];
-        if (methods) {
-            methods.forEach((m) => m.apply(this, invocationMessage.arguments));
-            if (invocationMessage.invocationId) {
-                // This is not supported in v1. So we return an error to avoid blocking the server waiting for the response.
-                const message = "Server requested a response, which is not supported in this version of the client.";
-                this.logger.log(LogLevel.Error, message);
-                this.connection.stop(new Error(message));
-            }
-        } else {
-            this.logger.log(LogLevel.Warning, `No client method with the name '${invocationMessage.target}' found.`);
-        }
-    }
-
-    private connectionClosed(error?: Error) {
-        const callbacks = this.callbacks;
-        this.callbacks = {};
-
-        Object.keys(callbacks)
-            .forEach((key) => {
-                const callback = callbacks[key];
-                callback(undefined, error ? error : new Error("Invocation canceled due to connection being closed."));
-            });
-
-        this.cleanupTimeout();
-
-        this.closedCallbacks.forEach((c) => c.apply(this, [error]));
-    }
-
     public async start(): Promise<void> {
         const handshakeRequest: HandshakeRequestMessage = {
             protocol: this.protocol.name,
             version: this.protocol.version,
         };
 
-        this.logger.log(LogLevel.Trace, "Starting HubConnection.");
+        this.logger.log(LogLevel.Debug, "Starting HubConnection.");
 
         this.receivedHandshakeResponse = false;
 
         await this.connection.start(this.protocol.transferFormat);
 
-        this.logger.log(LogLevel.Trace, "Sending handshake request.");
+        this.logger.log(LogLevel.Debug, "Sending handshake request.");
 
         await this.connection.send(this.handshakeProtocol.writeHandshakeRequest(handshakeRequest));
 
@@ -198,13 +85,13 @@ export class HubConnection {
     }
 
     public stop(): Promise<void> {
-        this.logger.log(LogLevel.Trace, "Stopping HubConnection.");
+        this.logger.log(LogLevel.Debug, "Stopping HubConnection.");
 
         this.cleanupTimeout();
         return this.connection.stop();
     }
 
-    public stream<T>(methodName: string, ...args: any[]): Observable<T> {
+    public stream<T = any>(methodName: string, ...args: any[]): IStreamResult<T> {
         const invocationDescriptor = this.createStreamInvocation(methodName, args);
 
         const subject = new Subject<T>(() => {
@@ -252,7 +139,7 @@ export class HubConnection {
         return this.connection.send(message);
     }
 
-    public invoke(methodName: string, ...args: any[]): Promise<any> {
+    public invoke<T = any>(methodName: string, ...args: any[]): Promise<T> {
         const invocationDescriptor = this.createInvocation(methodName, args, false);
 
         const p = new Promise<any>((resolve, reject) => {
@@ -327,12 +214,125 @@ export class HubConnection {
 
     }
 
-    public onclose(callback: ConnectionClosed) {
+    public onclose(callback: (error?: Error) => void) {
         if (callback) {
             this.closedCallbacks.push(callback);
         }
     }
 
+    private processIncomingData(data: any) {
+        this.cleanupTimeout();
+
+        if (!this.receivedHandshakeResponse) {
+            data = this.processHandshakeResponse(data);
+            this.receivedHandshakeResponse = true;
+        }
+
+        // Data may have all been read when processing handshake response
+        if (data) {
+            // Parse the messages
+            const messages = this.protocol.parseMessages(data, this.logger);
+
+            for (const message of messages) {
+                switch (message.type) {
+                    case MessageType.Invocation:
+                        this.invokeClientMethod(message);
+                        break;
+                    case MessageType.StreamItem:
+                    case MessageType.Completion:
+                        const callback = this.callbacks[message.invocationId];
+                        if (callback != null) {
+                            if (message.type === MessageType.Completion) {
+                                delete this.callbacks[message.invocationId];
+                            }
+                            callback(message);
+                        }
+                        break;
+                    case MessageType.Ping:
+                        // Don't care about pings
+                        break;
+                    case MessageType.Close:
+                        this.logger.log(LogLevel.Information, "Close message received from server.");
+                        this.connection.stop(message.error ? new Error("Server returned an error on close: " + message.error) : null);
+                        break;
+                    default:
+                        this.logger.log(LogLevel.Warning, "Invalid message type: " + message.type);
+                        break;
+                }
+            }
+        }
+
+        this.configureTimeout();
+    }
+
+    private processHandshakeResponse(data: any): any {
+        let responseMessage: HandshakeResponseMessage;
+        let remainingData: any;
+
+        try {
+            [remainingData, responseMessage] = this.handshakeProtocol.parseHandshakeResponse(data);
+        } catch (e) {
+            const message = "Error parsing handshake response: " + e;
+            this.logger.log(LogLevel.Error, message);
+
+            const error = new Error(message);
+            this.connection.stop(error);
+            throw error;
+        }
+        if (responseMessage.error) {
+            const message = "Server returned handshake error: " + responseMessage.error;
+            this.logger.log(LogLevel.Error, message);
+            this.connection.stop(new Error(message));
+        } else {
+            this.logger.log(LogLevel.Debug, "Server handshake complete.");
+        }
+
+        return remainingData;
+    }
+
+    private configureTimeout() {
+        if (!this.connection.features || !this.connection.features.inherentKeepAlive) {
+            // Set the timeout timer
+            this.timeoutHandle = setTimeout(() => this.serverTimeout(), this.timeoutInMilliseconds);
+        }
+    }
+
+    private serverTimeout() {
+        // The server hasn't talked to us in a while. It doesn't like us anymore ... :(
+        // Terminate the connection
+        this.connection.stop(new Error("Server timeout elapsed without receiving a message from the server."));
+    }
+
+    private invokeClientMethod(invocationMessage: InvocationMessage) {
+        const methods = this.methods[invocationMessage.target.toLowerCase()];
+        if (methods) {
+            methods.forEach((m) => m.apply(this, invocationMessage.arguments));
+            if (invocationMessage.invocationId) {
+                // This is not supported in v1. So we return an error to avoid blocking the server waiting for the response.
+                const message = "Server requested a response, which is not supported in this version of the client.";
+                this.logger.log(LogLevel.Error, message);
+                this.connection.stop(new Error(message));
+            }
+        } else {
+            this.logger.log(LogLevel.Warning, `No client method with the name '${invocationMessage.target}' found.`);
+        }
+    }
+
+    private connectionClosed(error?: Error) {
+        const callbacks = this.callbacks;
+        this.callbacks = {};
+
+        Object.keys(callbacks)
+            .forEach((key) => {
+                const callback = callbacks[key];
+                callback(undefined, error ? error : new Error("Invocation canceled due to connection being closed."));
+            });
+
+        this.cleanupTimeout();
+
+        this.closedCallbacks.forEach((c) => c.apply(this, [error]));
+    }
+
     private cleanupTimeout(): void {
         if (this.timeoutHandle) {
             clearTimeout(this.timeoutHandle);

clients/ts/signalr/src/IConnection.ts

@@ -1,16 +1,15 @@
 // 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 { ConnectionClosed, DataReceived } from "./Common";
 import { TransferFormat } from "./ITransport";
 
 export interface IConnection {
     readonly features: any;
 
     start(transferFormat: TransferFormat): Promise<void>;
-    send(data: any): Promise<void>;
+    send(data: string | ArrayBuffer): Promise<void>;
     stop(error?: Error): Promise<void>;
 
-    onreceive: DataReceived;
+    onreceive: (data: string | ArrayBuffer) => void;
-    onclose: ConnectionClosed;
+    onclose: (error?: Error) => void;
 }

clients/ts/signalr/src/ILogger.ts

@@ -1,12 +1,15 @@
 // 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.
 
+// These values are designed to match the ASP.NET Log Levels since that's the pattern we're emulating here.
 export enum LogLevel {
     Trace = 0,
-    Information,
+    Debug = 1,
-    Warning,
+    Information = 2,
-    Error,
+    Warning = 3,
-    None,
+    Error = 4,
+    Critical = 5,
+    None = 6,
 }
 
 export interface ILogger {

clients/ts/signalr/src/ITransport.ts

@@ -1,9 +1,8 @@
-import { DataReceived, TransportClosed } from "./Common";
-import { IConnection } from "./IConnection";
-
 // 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 { IConnection } from "./IConnection";
+
 export enum HttpTransportType {
     WebSockets,
     ServerSentEvents,
@@ -19,6 +18,6 @@ export interface ITransport {
     connect(url: string, transferFormat: TransferFormat): Promise<void>;
     send(data: any): Promise<void>;
     stop(): Promise<void>;
-    onreceive: DataReceived;
+    onreceive: (data: string | ArrayBuffer) => void;
-    onclose: TransportClosed;
+    onclose: (error?: Error) => void;
 }

clients/ts/signalr/src/JsonHubProtocol.ts

@@ -22,7 +22,7 @@ export class JsonHubProtocol implements IHubProtocol {
         }
 
         if (logger === null) {
-            logger = new NullLogger();
+            logger = NullLogger.instance;
         }
 
         // Parse the messages

clients/ts/signalr/src/Loggers.ts

@@ -4,51 +4,10 @@
 import { ILogger, LogLevel } from "./ILogger";
 
 export class NullLogger implements ILogger {
-    public log(logLevel: LogLevel, message: string): void {
+    public static instance: ILogger = new NullLogger();
-    }
-}
 
-export class ConsoleLogger implements ILogger {
+    private constructor() {}
-    private readonly minimumLogLevel: LogLevel;
-
-    constructor(minimumLogLevel: LogLevel) {
-        this.minimumLogLevel = minimumLogLevel;
-    }
 
     public log(logLevel: LogLevel, message: string): void {
-        if (logLevel >= this.minimumLogLevel) {
-            switch (logLevel) {
-                case LogLevel.Error:
-                    console.error(`${LogLevel[logLevel]}: ${message}`);
-                    break;
-                case LogLevel.Warning:
-                    console.warn(`${LogLevel[logLevel]}: ${message}`);
-                    break;
-                case LogLevel.Information:
-                    console.info(`${LogLevel[logLevel]}: ${message}`);
-                    break;
-                default:
-                    console.log(`${LogLevel[logLevel]}: ${message}`);
-                    break;
-            }
-        }
-    }
-}
-
-export class LoggerFactory {
-    public static createLogger(logging?: ILogger | LogLevel) {
-        if (logging === undefined) {
-            return new ConsoleLogger(LogLevel.Information);
-        }
-
-        if (logging === null) {
-            return new NullLogger();
-        }
-
-        if ((logging as ILogger).log) {
-            return logging as ILogger;
-        }
-
-        return new ConsoleLogger(logging as LogLevel);
     }
 }

clients/ts/signalr/src/LongPollingTransport.ts

@@ -2,7 +2,6 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 import { AbortController } from "./AbortController";
-import { DataReceived, TransportClosed } from "./Common";
 import { HttpError, TimeoutError } from "./Errors";
 import { HttpClient, HttpRequest } from "./HttpClient";
 import { ILogger, LogLevel } from "./ILogger";
@@ -188,6 +187,6 @@ export class LongPollingTransport implements ITransport {
         }
     }
 
-    public onreceive: DataReceived;
+    public onreceive: (data: string | ArrayBuffer) => void;
-    public onclose: TransportClosed;
+    public onclose: (error?: Error) => void;
 }

clients/ts/signalr/src/Observable.ts

@@ -1,73 +0,0 @@
-// 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.
-
-// TODO: Seamless RxJs integration
-// From RxJs: https://github.com/ReactiveX/rxjs/blob/master/src/Observer.ts
-export interface Observer<T> {
-    closed?: boolean;
-    next: (value: T) => void;
-    error?: (err: any) => void;
-    complete?: () => void;
-}
-
-export class Subscription<T> {
-    private subject: Subject<T>;
-    private observer: Observer<T>;
-
-    constructor(subject: Subject<T>, observer: Observer<T>) {
-        this.subject = subject;
-        this.observer = observer;
-    }
-
-    public dispose(): void {
-        const index: number = this.subject.observers.indexOf(this.observer);
-        if (index > -1) {
-            this.subject.observers.splice(index, 1);
-        }
-
-        if (this.subject.observers.length === 0) {
-            this.subject.cancelCallback().catch((_) => { });
-        }
-    }
-}
-
-export interface Observable<T> {
-    subscribe(observer: Observer<T>): Subscription<T>;
-}
-
-export class Subject<T> implements Observable<T> {
-    public observers: Array<Observer<T>>;
-    public cancelCallback: () => Promise<void>;
-
-    constructor(cancelCallback: () => Promise<void>) {
-        this.observers = [];
-        this.cancelCallback = cancelCallback;
-    }
-
-    public next(item: T): void {
-        for (const observer of this.observers) {
-            observer.next(item);
-        }
-    }
-
-    public error(err: any): void {
-        for (const observer of this.observers) {
-            if (observer.error) {
-                observer.error(err);
-            }
-        }
-    }
-
-    public complete(): void {
-        for (const observer of this.observers) {
-            if (observer.complete) {
-                observer.complete();
-            }
-        }
-    }
-
-    public subscribe(observer: Observer<T>): Subscription<T> {
-        this.observers.push(observer);
-        return new Subscription(this, observer);
-    }
-}

clients/ts/signalr/src/ServerSentEventsTransport.ts

@@ -1,7 +1,6 @@
 // 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 { DataReceived, TransportClosed } from "./Common";
 import { HttpClient } from "./HttpClient";
 import { ILogger, LogLevel } from "./ILogger";
 import { ITransport, TransferFormat } from "./ITransport";
@@ -106,6 +105,6 @@ export class ServerSentEventsTransport implements ITransport {
         }
     }
 
-    public onreceive: DataReceived;
+    public onreceive: (data: string | ArrayBuffer) => void;
-    public onclose: TransportClosed;
+    public onclose: (error?: Error) => void;
 }

clients/ts/signalr/src/Stream.ts

@@ -0,0 +1,23 @@
+// 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.
+
+// This is an API that is similar to Observable, but we don't want users to confuse it for that so we rename things. Someone could
+// easily adapt it into the Rx interface if they wanted to. Unlike in C#, we can't just implement an "interface" and get extension
+// methods for free. The methods have to actually be added to the object (there are no extension methods in JS!). We don't want to
+// depend on RxJS in the core library, so instead we duplicate the minimum logic needed and then users can easily adapt these into
+// proper RxJS observables if they want.
+
+export interface IStreamSubscriber<T> {
+    closed?: boolean;
+    next(value: T): void;
+    error(err: any): void;
+    complete(): void;
+}
+
+export interface IStreamResult<T> {
+    subscribe(observer: IStreamSubscriber<T>): ISubscription<T>;
+}
+
+export interface ISubscription<T> {
+    dispose(): void;
+}

clients/ts/signalr/src/Utils.ts

@@ -3,6 +3,8 @@
 
 import { HttpClient } from "./HttpClient";
 import { ILogger, LogLevel } from "./ILogger";
+import { NullLogger } from "./Loggers";
+import { IStreamResult, IStreamSubscriber, ISubscription } from "./Stream";
 
 export class Arg {
     public static isRequired(val: any, name: string): void {
@@ -67,3 +69,109 @@ export async function sendMessage(logger: ILogger, transportName: string, httpCl
 
     logger.log(LogLevel.Trace, `(${transportName} transport) request complete. Response status: ${response.statusCode}.`);
 }
+
+export function createLogger(logger?: ILogger | LogLevel) {
+    if (logger === undefined) {
+        return new ConsoleLogger(LogLevel.Information);
+    }
+
+    if (logger === null) {
+        return NullLogger.instance;
+    }
+
+    if ((logger as ILogger).log) {
+        return logger as ILogger;
+    }
+
+    return new ConsoleLogger(logger as LogLevel);
+}
+
+export class Subject<T> implements IStreamResult<T> {
+    public observers: Array<IStreamSubscriber<T>>;
+    public cancelCallback: () => Promise<void>;
+
+    constructor(cancelCallback: () => Promise<void>) {
+        this.observers = [];
+        this.cancelCallback = cancelCallback;
+    }
+
+    public next(item: T): void {
+        for (const observer of this.observers) {
+            observer.next(item);
+        }
+    }
+
+    public error(err: any): void {
+        for (const observer of this.observers) {
+            if (observer.error) {
+                observer.error(err);
+            }
+        }
+    }
+
+    public complete(): void {
+        for (const observer of this.observers) {
+            if (observer.complete) {
+                observer.complete();
+            }
+        }
+    }
+
+    public subscribe(observer: IStreamSubscriber<T>): ISubscription<T> {
+        this.observers.push(observer);
+        return new SubjectSubscription(this, observer);
+    }
+}
+
+export class SubjectSubscription<T> implements ISubscription<T> {
+    private subject: Subject<T>;
+    private observer: IStreamSubscriber<T>;
+
+    constructor(subject: Subject<T>, observer: IStreamSubscriber<T>) {
+        this.subject = subject;
+        this.observer = observer;
+    }
+
+    public dispose(): void {
+        const index: number = this.subject.observers.indexOf(this.observer);
+        if (index > -1) {
+            this.subject.observers.splice(index, 1);
+        }
+
+        if (this.subject.observers.length === 0) {
+            this.subject.cancelCallback().catch((_) => { });
+        }
+    }
+}
+
+export class ConsoleLogger implements ILogger {
+    private readonly minimumLogLevel: LogLevel;
+
+    constructor(minimumLogLevel: LogLevel) {
+        this.minimumLogLevel = minimumLogLevel;
+    }
+
+    public log(logLevel: LogLevel, message: string): void {
+        if (logLevel >= this.minimumLogLevel) {
+            switch (logLevel) {
+                case LogLevel.Critical:
+                case LogLevel.Error:
+                    console.error(`${LogLevel[logLevel]}: ${message}`);
+                    break;
+                case LogLevel.Warning:
+                    console.warn(`${LogLevel[logLevel]}: ${message}`);
+                    break;
+                case LogLevel.Information:
+                    console.info(`${LogLevel[logLevel]}: ${message}`);
+                    break;
+                case LogLevel.Trace:
+                case LogLevel.Debug:
+                    console.debug(`${LogLevel[logLevel]}: ${message}`);
+                    break;
+                default:
+                    console.log(`${LogLevel[logLevel]}: ${message}`);
+                    break;
+            }
+        }
+    }
+}

clients/ts/signalr/src/WebSocketTransport.ts

@@ -1,7 +1,6 @@
 // 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 { DataReceived, TransportClosed } from "./Common";
 import { ILogger, LogLevel } from "./ILogger";
 import { ITransport, TransferFormat } from "./ITransport";
 import { Arg, getDataDetail } from "./Utils";
@@ -90,6 +89,6 @@ export class WebSocketTransport implements ITransport {
         return Promise.resolve();
     }
 
-    public onreceive: DataReceived;
+    public onreceive: (data: string | ArrayBuffer) => void;
-    public onclose: TransportClosed;
+    public onclose: (error?: Error) => void;
 }

clients/ts/signalr/src/index.ts

@@ -2,7 +2,6 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 // Everything that users need to access must be exported here. Including interfaces.
-export * from "./Common";
 export * from "./Errors";
 export * from "./HttpClient";
 export * from "./HttpConnection";
@@ -10,6 +9,6 @@ export * from "./HubConnection";
 export * from "./IConnection";
 export * from "./IHubProtocol";
 export * from "./ILogger";
-export * from "./Loggers";
 export * from "./ITransport";
-export * from "./Observable";
+export * from "./Stream";
+export * from "./Loggers";

samples/ChatSample/Views/Home/Index.cshtml

@@ -16,9 +16,8 @@
 </div>
 <script src="lib/signalr-client/signalr.js"></script>
 <script>
-let transportType = signalR.TransportType[getParameterByName('transport')] || signalR.TransportType.WebSockets;
+let transportType = signalR.HttpTransportType[getParameterByName('transport')] || signalR.HttpTransportType.WebSockets;
-let logger = new signalR.ConsoleLogger(signalR.LogLevel.Information);
+let connection = new signalR.HubConnection('/chat', { transport: transportType, logger: signalR.LogLevel.Information });
-let connection = new signalR.HubConnection('/chat', { transport: transportType, logger: logger });
 
 connection.onclose(e => {
     if (e) {

samples/ChatSample/web.config

@@ -1,14 +1,14 @@
 <?xml version="1.0" encoding="utf-8"?>
 <configuration>
-
   <!--
     Configure your application settings in appsettings.json. Learn more at http://go.microsoft.com/fwlink/?LinkId=786380
   -->
-
   <system.webServer>
     <handlers>
-      <add name="aspNetCore" path="*" verb="*" modules="AspNetCoreModule" resourceType="Unspecified"/>
+      <add name="aspNetCore" path="*" verb="*" modules="AspNetCoreModule" resourceType="Unspecified" />
     </handlers>
-    <aspNetCore processPath="%LAUNCHER_PATH%" arguments="%LAUNCHER_ARGS%" stdoutLogEnabled="false" stdoutLogFile=".\logs\stdout" forwardWindowsAuthToken="false"/>
+    <aspNetCore processPath="%LAUNCHER_PATH%" arguments="%LAUNCHER_ARGS%" stdoutLogEnabled="false" stdoutLogFile=".\logs\stdout" forwardWindowsAuthToken="false" startupTimeLimit="3600" requestTimeout="23:00:00">
+      <environmentVariables />
+    </aspNetCore>
   </system.webServer>
-</configuration>
+</configuration>

samples/JwtSample/wwwroot/index.html

@@ -1,4 +1,4 @@
-<!DOCTYPE html>
+<!DOCTYPE html>
 <html>
 <head>
     <meta charset="utf-8" />
@@ -94,8 +94,8 @@
             });
     }
 
-    [signalR.TransportType.WebSockets, signalR.TransportType.ServerSentEvents, signalR.TransportType.LongPolling].forEach(function(transportType) {
+    [signalR.HttpTransportType.WebSockets, signalR.HttpTransportType.ServerSentEvents, signalR.HttpTransportType.LongPolling].forEach(function(transportType) {
-        var clientId = 'browser ' + signalR.TransportType[transportType];
+        var clientId = 'browser ' + signalR.HttpTransportType[transportType];
         createLog(clientId);
         appendLog(clientId, 'Log for user: ' + clientId);
         runConnection(clientId, transportType);

samples/SignalRSamples/wwwroot/hubs.html

@@ -103,8 +103,6 @@
         return document.getElementById(id).value;
     }
 
-    let logger = new signalR.ConsoleLogger(signalR.LogLevel.Trace);
-
     let connectButton = document.getElementById('connect');
     let disconnectButton = document.getElementById('disconnect');
     let broadcastButton = document.getElementById('broadcast');
@@ -141,10 +139,10 @@
             new signalR.protocols.msgpack.MessagePackHubProtocol() :
             new signalR.JsonHubProtocol();
 
-        let options = { logger: logger, protocol: protocol };
+        let options = { logger: signalR.LogLevel.Trace, protocol: protocol };
 
         if (transportDropdown.value !== "Automatic") {
-            options.transport = signalR.TransportType[transportDropdown.value];
+            options.transport = signalR.HttpTransportType[transportDropdown.value];
         }
 
         console.log('http://' + document.location.host + '/' + hubRoute);

samples/SignalRSamples/wwwroot/sockets.html

@@ -19,12 +19,12 @@
     <script src="utils.js"></script>
     <script>
         document.addEventListener('DOMContentLoaded', function () {
-            let transportType = signalR.TransportType[getParameterByName('transport')] || signalR.TransportType.WebSockets;
+            let transportType = signalR.HttpTransportType[getParameterByName('transport')] || signalR.HttpTransportType.WebSockets;
 
-            document.getElementById('transportName').innerHTML = signalR.TransportType[transportType];
+            document.getElementById('transportName').innerHTML = signalR.HttpTransportType[transportType];
 
             let url = 'http://' + document.location.host + '/chat';
-            let connection = new signalR.HttpConnection(url, { transport: transportType, logger: new signalR.ConsoleLogger(signalR.LogLevel.Trace) });
+            let connection = new signalR.HttpConnection(url, { transport: transportType, logger: signalR.LogLevel.Trace });
 
             connection.onreceive = function (data) {
                 let child = document.createElement('li');

samples/SignalRSamples/wwwroot/streaming.html

@@ -37,12 +37,11 @@
             let connectButton = document.getElementById('connectButton');
             let disconnectButton = document.getElementById('disconnectButton');
 
-            let logger = new signalR.ConsoleLogger(signalR.LogLevel.Trace);
+            let transportType = signalR.HttpTransportType[getParameterByName('transport')] || signalR.HttpTransportType.WebSockets;
-            let transportType = signalR.TransportType[getParameterByName('transport')] || signalR.TransportType.WebSockets;
 
             let invocationCounter = 0;
 
-            document.getElementById('transportName').innerHTML = signalR.TransportType[transportType];
+            document.getElementById('transportName').innerHTML = signalR.HttpTransportType[transportType];
             let connection = null;
 
             click('clearButton', function () {
@@ -54,7 +53,7 @@
             });
 
             click('connectButton', function () {
-                connection = new signalR.HubConnection('/streaming', { transport: transportType, logger: logger });
+                connection = new signalR.HubConnection('/streaming', { transport: transportType, logger: signalR.LogLevel.Trace });
 
                 connection.onclose(function () {
                     channelButton.disabled = true;
@@ -76,11 +75,11 @@
             });
 
             click('observableButton', function () {
-                run('ObservableCounter')
+                run('ObservableCounter');
             });
 
             click('channelButton', function () {
-                run('ChannelCounter')
+                run('ChannelCounter');
             });
 
             function run(method) {