Merge branch 'release/2.1' into dev

.gitignore

@@ -42,4 +42,5 @@ site.min.css
 dist/
 global.json
 BenchmarkDotNet.Artifacts/
-.rpt2_cache/
+.rpt2_cache/
+*.orig

client-ts/FunctionalTests/ts/HubConnectionTests.ts

@@ -345,12 +345,6 @@ describe("hubConnection", () => {
             });
 
             it("closed with error if hub cannot be created", (done) => {
-                const errorRegex = {
-                    LongPolling: "Internal Server Error",
-                    ServerSentEvents: "Error occurred",
-                    WebSockets: "1011|1005", // Message is browser specific (e.g. 'Websocket closed with status code: 1011'), Edge and IE report 1005 even though the server sent 1011
-                };
-
                 const hubConnection = new HubConnection("http://" + document.location.host + "/uncreatable", {
                     logger: LogLevel.Trace,
                     protocol,
@@ -358,7 +352,7 @@ describe("hubConnection", () => {
                 });
 
                 hubConnection.onclose((error) => {
-                    expect(error.message).toMatch(errorRegex[TransportType[transportType]]);
+                    expect(error.message).toEqual("Server returned an error on close: Connection closed with an error. InvalidOperationException: Unable to resolve service for type 'System.Object' while attempting to activate 'FunctionalTests.UncreatableHub'.");
                     done();
                 });
                 hubConnection.start();

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

@@ -38,11 +38,25 @@ export class MessagePackHubProtocol implements IHubProtocol {
                 return this.createCompletionMessage(this.readHeaders(properties), properties);
             case MessageType.Ping:
                 return this.createPingMessage(properties);
+            case MessageType.Close:
+                return this.createCloseMessage(properties);
             default:
                 throw new Error("Invalid message type.");
         }
     }
 
+    private createCloseMessage(properties: any[]): HubMessage {
+        if (properties.length !== 2) {
+            throw new Error("Invalid payload for Close message.");
+        }
+
+        return {
+            // Close messages have no headers.
+            error: properties[1],
+            type: MessageType.Close,
+        } as HubMessage;
+    }
+
     private createPingMessage(properties: any[]): HubMessage {
         if (properties.length !== 1) {
             throw new Error("Invalid payload for Ping message.");

client-ts/signalr/spec/HubConnection.spec.ts

@@ -4,7 +4,7 @@
 import { ConnectionClosed, DataReceived } from "../src/Common";
 import { HubConnection } from "../src/HubConnection";
 import { IConnection } from "../src/IConnection";
-import { MessageType } from "../src/IHubProtocol";
+import { MessageType, IHubProtocol, HubMessage } from "../src/IHubProtocol";
 import { ILogger, LogLevel } from "../src/ILogger";
 import { Observer } from "../src/Observable";
 import { TextMessageFormat } from "../src/TextMessageFormat";
@@ -80,10 +80,91 @@ describe("HubConnection", () => {
             hubConnection.stop();
         });
 
+        it("can process handshake from text", async () => {
+            let protocolCalled = false;
+
+            const mockProtocol = new TestProtocol(TransferFormat.Text);
+            mockProtocol.onreceive = (d) => {
+                protocolCalled = true;
+            };
+
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, { logger: null, protocol: mockProtocol });
+
+            const data = "{}" + TextMessageFormat.RecordSeparator;
+
+            connection.receiveText(data);
+
+            // message only contained handshake response
+            expect(protocolCalled).toEqual(false);
+        });
+
+        it("can process handshake from binary", async () => {
+            let protocolCalled = false;
+
+            const mockProtocol = new TestProtocol(TransferFormat.Binary);
+            mockProtocol.onreceive = (d) => {
+                protocolCalled = true;
+            };
+
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, { logger: null, protocol: mockProtocol });
+
+            // handshake response + message separator
+            const data = [0x7b, 0x7d, 0x1e];
+
+            connection.receiveBinary(new Uint8Array(data).buffer);
+
+            // message only contained handshake response
+            expect(protocolCalled).toEqual(false);
+        });
+
+        it("can process handshake and additional messages from binary", async () => {
+            let receivedProcotolData: ArrayBuffer;
+
+            const mockProtocol = new TestProtocol(TransferFormat.Binary);
+            mockProtocol.onreceive = (d) => receivedProcotolData = d;
+
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, { logger: null, protocol: mockProtocol });
+
+            // handshake response + message separator + message pack message
+            const data = [
+                0x7b, 0x7d, 0x1e, 0x65, 0x95, 0x03, 0x80, 0xa1, 0x30, 0x01, 0xd9, 0x5d, 0x54, 0x68, 0x65, 0x20, 0x63, 0x6c,
+                0x69, 0x65, 0x6e, 0x74, 0x20, 0x61, 0x74, 0x74, 0x65, 0x6d, 0x70, 0x74, 0x65, 0x64, 0x20, 0x74, 0x6f, 0x20,
+                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
+            ];
+
+            connection.receiveBinary(new Uint8Array(data).buffer);
+
+            // left over data is the message pack message
+            expect(receivedProcotolData.byteLength).toEqual(102);
+        });
+
+        it("can process handshake and additional messages from text", async () => {
+            let receivedProcotolData: string;
+
+            const mockProtocol = new TestProtocol(TransferFormat.Text);
+            mockProtocol.onreceive = (d) => receivedProcotolData = d;
+
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, { logger: null, protocol: mockProtocol });
+
+            const data = "{}" + TextMessageFormat.RecordSeparator + "{\"type\":6}" + TextMessageFormat.RecordSeparator;
+
+            connection.receiveText(data);
+
+            expect(receivedProcotolData).toEqual("{\"type\":6}" + TextMessageFormat.RecordSeparator);
+        });
+
         it("rejects the promise when an error is received", async () => {
             const connection = new TestConnection();
-
             const hubConnection = new HubConnection(connection, commonOptions);
+            connection.receiveHandshakeResponse();
+
             const invokePromise = hubConnection.invoke("testMethod", "arg", 42);
 
             connection.receive({ type: MessageType.Completion, invocationId: connection.lastInvocationId, error: "foo" });
@@ -94,8 +175,9 @@ describe("HubConnection", () => {
 
         it("resolves the promise when a result is received", async () => {
             const connection = new TestConnection();
-
             const hubConnection = new HubConnection(connection, commonOptions);
+            connection.receiveHandshakeResponse();
+
             const invokePromise = hubConnection.invoke("testMethod", "arg", 42);
 
             connection.receive({ type: MessageType.Completion, invocationId: connection.lastInvocationId, result: "foo" });
@@ -107,6 +189,9 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
 
             const hubConnection = new HubConnection(connection, commonOptions);
+
+            connection.receiveHandshakeResponse();
+
             const invokePromise = hubConnection.invoke("testMethod");
             hubConnection.stop();
 
@@ -118,6 +203,9 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
 
             const hubConnection = new HubConnection(connection, commonOptions);
+    
+            connection.receiveHandshakeResponse();
+
             const invokePromise = hubConnection.invoke("testMethod");
             // Typically this would be called by the transport
             connection.onclose(new Error("Connection lost"));
@@ -140,6 +228,8 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, { logger });
 
+            connection.receiveHandshakeResponse();
+
             connection.receive({
                 arguments: ["test"],
                 invocationId: 0,
@@ -163,6 +253,8 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, { logger });
 
+            connection.receiveHandshakeResponse();
+
             const handler = () => { };
             hubConnection.on('message', handler);
             hubConnection.off('message', handler);
@@ -181,6 +273,9 @@ describe("HubConnection", () => {
         it("callback invoked when servers invokes a method on the client", async () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, commonOptions);
+ 
+            connection.receiveHandshakeResponse();
+
             let value = "";
             hubConnection.on("message", (v) => value = v);
 
@@ -195,9 +290,67 @@ describe("HubConnection", () => {
             expect(value).toBe("test");
         });
 
+        it("stop on handshake error", async () => {
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, commonOptions);
+   
+            let closeError: Error = null;
+            hubConnection.onclose((e) => closeError = e);
+
+            connection.receiveHandshakeResponse("Error!");
+
+            expect(closeError.message).toEqual("Server returned handshake error: Error!");
+        });
+
+        it("stop on close message", async () => {
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, commonOptions);
+   
+            let isClosed = false;
+            let closeError: Error = null;
+            hubConnection.onclose((e) => {
+                isClosed = true;
+                closeError = e;
+            });
+
+            connection.receiveHandshakeResponse();
+
+            connection.receive({
+                type: MessageType.Close,
+            });
+
+            expect(isClosed).toEqual(true);
+            expect(closeError).toEqual(null);
+        });
+
+        it("stop on error close message", async () => {
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, commonOptions);
+   
+            let isClosed = false;
+            let closeError: Error = null;
+            hubConnection.onclose((e) => {
+                isClosed = true;
+                closeError = e;
+            });
+
+            connection.receiveHandshakeResponse();
+
+            connection.receive({
+                error: "Error!",
+                type: MessageType.Close,
+            });
+
+            expect(isClosed).toEqual(true);
+            expect(closeError.message).toEqual("Server returned an error on close: Error!");
+        });
+
         it("can have multiple callbacks", async () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, commonOptions);
+   
+            connection.receiveHandshakeResponse();
+
             let numInvocations1 = 0;
             let numInvocations2 = 0;
             hubConnection.on("message", () => numInvocations1++);
@@ -219,6 +372,8 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, commonOptions);
 
+            connection.receiveHandshakeResponse();
+
             let numInvocations = 0;
             const callback = () => numInvocations++;
             hubConnection.on("message", callback);
@@ -267,6 +422,8 @@ describe("HubConnection", () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, { logger });
 
+            connection.receiveHandshakeResponse();
+
             hubConnection.on(null, undefined);
             hubConnection.on(undefined, null);
             hubConnection.on("message", null);
@@ -319,8 +476,10 @@ describe("HubConnection", () => {
 
         it("completes with an error when an error is yielded", async () => {
             const connection = new TestConnection();
-
             const hubConnection = new HubConnection(connection, commonOptions);
+
+            connection.receiveHandshakeResponse();
+
             const observer = new TestObserver();
             hubConnection.stream<any>("testMethod", "arg", 42)
                 .subscribe(observer);
@@ -333,8 +492,10 @@ describe("HubConnection", () => {
 
         it("completes the observer when a completion is received", async () => {
             const connection = new TestConnection();
-
             const hubConnection = new HubConnection(connection, commonOptions);
+      
+            connection.receiveHandshakeResponse();
+
             const observer = new TestObserver();
             hubConnection.stream<any>("testMethod", "arg", 42)
                 .subscribe(observer);
@@ -374,8 +535,10 @@ describe("HubConnection", () => {
 
         it("yields items as they arrive", async () => {
             const connection = new TestConnection();
-
             const hubConnection = new HubConnection(connection, commonOptions);
+
+            connection.receiveHandshakeResponse();
+
             const observer = new TestObserver();
             hubConnection.stream<any>("testMethod")
                 .subscribe(observer);
@@ -421,8 +584,10 @@ describe("HubConnection", () => {
 
         it("can be canceled", () => {
             const connection = new TestConnection();
-
             const hubConnection = new HubConnection(connection, commonOptions);
+
+            connection.receiveHandshakeResponse();
+
             const observer = new TestObserver();
             const subscription = hubConnection.stream("testMethod")
                 .subscribe(observer);
@@ -518,6 +683,33 @@ describe("HubConnection", () => {
             expect(error).toBeUndefined();
         });
 
+        it("does not timeout if message was received before HubConnection.start", async () => {
+            const connection = new TestConnection();
+            const hubConnection = new HubConnection(connection, { ...commonOptions, timeoutInMilliseconds: 100 });
+
+            const p = new PromiseSource<Error>();
+            hubConnection.onclose((e) => p.resolve(e));
+
+            // send message before start to trigger timeout handler
+            // testing for regression where we didn't cleanup timer if request received before start created a timer
+            await connection.receive({ type: MessageType.Ping });
+
+            await hubConnection.start();
+
+            await connection.receive({ type: MessageType.Ping });
+            await delay(50);
+            await connection.receive({ type: MessageType.Ping });
+            await delay(50);
+            await connection.receive({ type: MessageType.Ping });
+            await delay(50);
+
+            connection.stop();
+
+            const error = await p.promise;
+
+            expect(error).toBeUndefined();
+        });
+
         it("terminates if no messages received within timeout interval", async () => {
             const connection = new TestConnection();
             const hubConnection = new HubConnection(connection, { ...commonOptions, timeoutInMilliseconds: 100 });
@@ -562,17 +754,53 @@ class TestConnection implements IConnection {
         return Promise.resolve();
     }
 
+    public receiveHandshakeResponse(error?: string): void {
+        this.receive({error: error});
+    }
+
     public receive(data: any): void {
         const payload = JSON.stringify(data);
         this.onreceive(TextMessageFormat.write(payload));
     }
 
+    public receiveText(data: string) {
+        this.onreceive(data);
+    }
+
+    public receiveBinary(data: ArrayBuffer) {
+        this.onreceive(data);
+    }
+
     public onreceive: DataReceived;
     public onclose: ConnectionClosed;
     public sentData: any[];
     public lastInvocationId: string;
 }
 
+class TestProtocol implements IHubProtocol {
+    public readonly name: string = "TestProtocol";
+
+    public readonly transferFormat: TransferFormat;
+
+    public onreceive: DataReceived;
+
+    constructor(transferFormat: TransferFormat) {
+        this.transferFormat = transferFormat;
+    }
+
+    public parseMessages(input: any): HubMessage[] {
+        if (this.onreceive) {
+            this.onreceive(input);
+        }
+
+        return [];
+    }
+
+    public writeMessage(message: HubMessage): any {
+
+    }
+}
+
 class TestObserver implements Observer<any> {
     public itemsReceived: [any];
     private itemsSource: PromiseSource<[any]>;

client-ts/signalr/spec/tsconfig.json

@@ -2,10 +2,11 @@
     "compileOnSave": false,
     "compilerOptions": {
         "module": "commonjs",
-        "target": "es5",
+        "target": "es2016",
+        "sourceMap": true,
         "moduleResolution": "node",
         "outDir": "./obj",
-        "lib": [ "es2015", "dom" ]
+        "lib": [ "es2016", "dom" ]
     },
     "include": [
         "./**/*",

client-ts/signalr/src/HttpClient.ts

@@ -54,6 +54,7 @@ export class DefaultHttpClient extends HttpClient {
             const xhr = new XMLHttpRequest();
 
             xhr.open(request.method, request.url, true);
+            xhr.withCredentials = true;
             xhr.setRequestHeader("X-Requested-With", "XMLHttpRequest");
 
             if (request.headers) {

client-ts/signalr/src/HubConnection.ts

@@ -4,7 +4,7 @@
 import { ConnectionClosed } from "./Common";
 import { HttpConnection, IHttpConnectionOptions } from "./HttpConnection";
 import { IConnection } from "./IConnection";
-import { CancelInvocationMessage, CompletionMessage, HubMessage, IHubProtocol, InvocationMessage, MessageType, NegotiationMessage, StreamInvocationMessage, StreamItemMessage } from "./IHubProtocol";
+import { CancelInvocationMessage, CompletionMessage, HandshakeRequestMessage, HandshakeResponseMessage, HubMessage, IHubProtocol, InvocationMessage, MessageType, StreamInvocationMessage, StreamItemMessage } from "./IHubProtocol";
 import { ILogger, LogLevel } from "./ILogger";
 import { JsonHubProtocol } from "./JsonHubProtocol";
 import { ConsoleLogger, LoggerFactory, NullLogger } from "./Loggers";
@@ -31,6 +31,7 @@ export class HubConnection {
     private closedCallbacks: ConnectionClosed[];
     private timeoutHandle: NodeJS.Timer;
     private timeoutInMilliseconds: number;
+    private receivedHandshakeResponse: boolean;
 
     constructor(url: string, options?: IHubConnectionOptions);
     constructor(connection: IConnection, options?: IHubConnectionOptions);
@@ -59,40 +60,106 @@ export class HubConnection {
     }
 
     private processIncomingData(data: any) {
-        if (this.timeoutHandle !== undefined) {
+        this.cleanupTimeout();
-            clearTimeout(this.timeoutHandle);
+
+        if (!this.receivedHandshakeResponse) {
+            data = this.processHandshakeResponse(data);
+            this.receivedHandshakeResponse = true;
         }
 
-        // Parse the messages
+        // Data may have all been read when processing handshake response
-        const messages = this.protocol.parseMessages(data);
+        if (data) {
+            // Parse the messages
+            const messages = this.protocol.parseMessages(data);
 
-        for (const message of messages) {
+            for (const message of messages) {
-            switch (message.type) {
+                switch (message.type) {
-                case MessageType.Invocation:
+                    case MessageType.Invocation:
-                    this.invokeClientMethod(message);
+                        this.invokeClientMethod(message);
-                    break;
+                        break;
-                case MessageType.StreamItem:
+                    case MessageType.StreamItem:
-                case MessageType.Completion:
+                    case MessageType.Completion:
-                    const callback = this.callbacks.get(message.invocationId);
+                        const callback = this.callbacks.get(message.invocationId);
-                    if (callback != null) {
+                        if (callback != null) {
-                        if (message.type === MessageType.Completion) {
+                            if (message.type === MessageType.Completion) {
-                            this.callbacks.delete(message.invocationId);
+                                this.callbacks.delete(message.invocationId);
+                            }
+                            callback(message);
                         }
-                        callback(message);
+                        break;
-                    }
+                    case MessageType.Ping:
-                    break;
+                        // Don't care about pings
-                case MessageType.Ping:
+                        break;
-                    // Don't care about pings
+                    case MessageType.Close:
-                    break;
+                        this.logger.log(LogLevel.Information, "Close message received from server.");
-                default:
+                        this.connection.stop(message.error ? new Error("Server returned an error on close: " + message.error) : null);
-                    this.logger.log(LogLevel.Warning, "Invalid message type: " + data);
+                        break;
-                    break;
+                    default:
+                        this.logger.log(LogLevel.Warning, "Invalid message type: " + data);
+                        break;
+                }
             }
         }
 
         this.configureTimeout();
     }
 
+    private processHandshakeResponse(data: any): any {
+        let responseMessage: HandshakeResponseMessage;
+        let messageData: string;
+        let remainingData: any;
+        try {
+            if (data instanceof ArrayBuffer) {
+                // Format is binary but still need to read JSON text from handshake response
+                const binaryData = new Uint8Array(data);
+                const separatorIndex = binaryData.indexOf(TextMessageFormat.RecordSeparatorCode);
+                if (separatorIndex === -1) {
+                    throw new Error("Message is incomplete.");
+                }
+
+                // content before separator is handshake response
+                // optional content after is additional messages
+                const responseLength = separatorIndex + 1;
+                messageData = String.fromCharCode.apply(null, binaryData.slice(0, responseLength));
+                remainingData = (binaryData.byteLength > responseLength) ? binaryData.slice(responseLength).buffer : null;
+            } else {
+                const textData: string = data;
+                const separatorIndex = textData.indexOf(TextMessageFormat.RecordSeparator);
+                if (separatorIndex === -1) {
+                    throw new Error("Message is incomplete.");
+                }
+
+                // content before separator is handshake response
+                // optional content after is additional messages
+                const responseLength = separatorIndex + 1;
+                messageData = textData.substring(0, responseLength);
+                remainingData = (textData.length > responseLength) ? textData.substring(responseLength) : null;
+            }
+
+            // At this point we should have just the single handshake message
+            const messages = TextMessageFormat.parse(messageData);
+            responseMessage = JSON.parse(messages[0]);
+        } 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.");
+        }
+
+        // multiple messages could have arrived with handshake
+        // return additional data to be parsed as usual, or null if all parsed
+        return remainingData;
+    }
+
     private configureTimeout() {
         if (!this.connection.features || !this.connection.features.inherentKeepAlive) {
             // Set the timeout timer
@@ -127,20 +194,25 @@ export class HubConnection {
         });
         this.callbacks.clear();
 
-        this.closedCallbacks.forEach((c) => c.apply(this, [error]));
-
         this.cleanupTimeout();
+
+        this.closedCallbacks.forEach((c) => c.apply(this, [error]));
     }
 
     public async start(): Promise<void> {
+        this.receivedHandshakeResponse = false;
+
         await this.connection.start(this.protocol.transferFormat);
 
+        // Handshake request is always JSON
         await this.connection.send(
             TextMessageFormat.write(
-                JSON.stringify({ protocol: this.protocol.name } as NegotiationMessage)));
+                JSON.stringify({ protocol: this.protocol.name } as HandshakeRequestMessage)));
 
         this.logger.log(LogLevel.Information, `Using HubProtocol '${this.protocol.name}'.`);
 
+        // defensively cleanup timeout in case we receive a message from the server before we finish start
+        this.cleanupTimeout();
         this.configureTimeout();
     }
 

client-ts/signalr/src/IHubProtocol.ts

@@ -10,11 +10,12 @@ export const enum MessageType {
     StreamInvocation = 4,
     CancelInvocation = 5,
     Ping = 6,
+    Close = 7,
 }
 
 export interface MessageHeaders { [key: string]: string; }
 
-export type HubMessage = InvocationMessage | StreamInvocationMessage | StreamItemMessage | CompletionMessage | CancelInvocationMessage | PingMessage;
+export type HubMessage = InvocationMessage | StreamInvocationMessage | StreamItemMessage | CompletionMessage | CancelInvocationMessage | PingMessage | CloseMessage;
 
 export interface HubMessageBase {
     readonly type: MessageType;
@@ -48,14 +49,23 @@ export interface CompletionMessage extends HubInvocationMessage {
     readonly result?: any;
 }
 
-export interface NegotiationMessage {
+export interface HandshakeRequestMessage {
     readonly protocol: string;
 }
 
-export interface PingMessage extends HubInvocationMessage {
+export interface HandshakeResponseMessage {
+    readonly error: string;
+}
+
+export interface PingMessage extends HubMessageBase {
     readonly type: MessageType.Ping;
 }
 
+export interface CloseMessage extends HubMessageBase {
+    readonly type: MessageType.Close;
+    readonly error?: string;
+}
+
 export interface CancelInvocationMessage extends HubInvocationMessage {
     readonly type: MessageType.CancelInvocation;
 }

client-ts/signalr/src/TextMessageFormat.ts

@@ -2,7 +2,8 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 export class TextMessageFormat {
-    private static RecordSeparator = String.fromCharCode(0x1e);
+    public static RecordSeparatorCode = 0x1e;
+    public static RecordSeparator = String.fromCharCode(TextMessageFormat.RecordSeparatorCode);
 
     public static write(output: string): string {
         return `${output}${TextMessageFormat.RecordSeparator}`;

client-ts/signalr/src/Transports.ts

@@ -145,7 +145,7 @@ export class ServerSentEventsTransport implements ITransport {
                 url += (url.indexOf("?") < 0 ? "?" : "&") + `access_token=${encodeURIComponent(token)}`;
             }
 
-            const eventSource = new EventSource(url);
+            const eventSource = new EventSource(url, { withCredentials: true });
 
             try {
                 eventSource.onmessage = (e: MessageEvent) => {

samples/SocketsSample/Startup.cs

@@ -38,7 +38,8 @@ namespace SocketsSample
                 {
                     p.AllowAnyHeader()
                      .AllowAnyMethod()
-                     .AllowAnyOrigin();
+                     .AllowAnyOrigin()
+                     .AllowCredentials();
                 });
             });
 

specs/HubProtocol.md

@@ -4,8 +4,8 @@ The SignalR Protocol is a protocol for two-way RPC over any Message-based transp
 
 ## Terms
 
-* Caller - The node that is issuing an `Negotiation`, `Invocation`, `StreamInvocation`, `CancelInvocation` messages and receiving `Completion` and `StreamItem` messages (a node can be both Caller and Callee for different invocations simultaneously)
+* Caller - The node that is issuing an `Invocation`, `StreamInvocation`, `CancelInvocation`, `Ping` messages and receiving `Completion`, `StreamItem` and `Ping` messages (a node can be both Caller and Callee for different invocations simultaneously)
-* Callee - The node that is receiving an `Negotiation`, `Invocation`, `StreamInvocation`, `CancelInvocation` messages and issuing `Completion` and `StreamItem` messages (a node can be both Callee and Caller for different invocations simultaneously)
+* Callee - The node that is receiving an `Invocation`, `StreamInvocation`, `CancelInvocation`, `Ping` messages and issuing `Completion`, `StreamItem` and `Ping` messages (a node can be both Callee and Caller for different invocations simultaneously)
 * Binder - The component on each node that handles mapping `Invocation` and `StreamInvocation` messages to method calls and return values to `Completion` and `StreamItem` messages
 
 ## Transport Requirements
@@ -16,21 +16,25 @@ The SignalR Protocol requires the following attributes from the underlying trans
 
 ## Overview
 
-This document describes two encodings of the SignalR protocol: [JSON](http://www.json.org/) and [MessagePack](http://msgpack.org/). Only one format can be used for the duration of a connection, and the format must be negotiated after opening the connection and before sending any other messages. However, each format shares a similar overall structure.
+This document describes two encodings of the SignalR protocol: [JSON](http://www.json.org/) and [MessagePack](http://msgpack.org/). Only one format can be used for the duration of a connection, and the format must be agreed on by both sides after opening the connection and before sending any other messages. However, each format shares a similar overall structure.
 
 In the SignalR protocol, the following types of messages can be sent:
 
-* `Negotiation` Message - Sent by the client to negotiate the message format.
+| Message Name          | Sender         | Description                                                                                                                    |
-* `Invocation` Message - Indicates a request to invoke a particular method (the Target) with provided Arguments on the remote endpoint.
+| ------------------    | -------------- | ------------------------------------------------------------------------------------------------------------------------------ |
-* `StreamInvocation` Message - Indicates a request to invoke a streaming method (the Target) with provided Arguments on the remote endpoint.
+| `HandshakeRequest`    | Client         | Sent by the client to agree on the message format.                                                                            |
-* `StreamItem` Message - Indicates individual items of streamed response data from a previous Invocation message.
+| `HandshakeResponse`   | Server         | Sent by the server as an acknowledgment of the previous `HandshakeRequest` message. Contains an error if the handshake failed. |
-* `Completion` Message - Indicates a previous Invocation or StreamInvocation has completed. Contains an error if the invocation concluded with an error or the result of a non-streaming method invocation. The result will be absent for `void` methods. In case of streaming invocations no further `StreamItem` messages will be received
+| `Close`               | Callee, Caller | Sent by the server when a connection is closed. Contains an error if the connection was closed because of an error.            |
-* `CancelInvocation` Message - Sent by the client to cancel a streaming invocation on the server.
+| `Invocation`          | Caller         | Indicates a request to invoke a particular method (the Target) with provided Arguments on the remote endpoint.                 |
-* `Ping` Message - Sent by either party to check if the connection is active.
+| `StreamInvocation`    | Caller         | Indicates a request to invoke a streaming method (the Target) with provided Arguments on the remote endpoint.                  |
+| `StreamItem`          | Callee         | Indicates individual items of streamed response data from a previous `StreamInvocation` message.                               |
+| `Completion`          | Callee         | Indicates a previous `Invocation` or `StreamInvocation` has completed. Contains an error if the invocation concluded with an error or the result of a non-streaming method invocation. The result will be absent for `void` methods. In case of streaming invocations no further `StreamItem` messages will be received. |
+| `CancelInvocation`    | Caller         | Sent by the client to cancel a streaming invocation on the server.                                                             |
+| `Ping`                | Caller, Callee | Sent by either party to check if the connection is active.                                                                     |
 
-After opening a connection to the server the client must send a `Negotiation` message to the server as its first message. The negotiation message is **always** a JSON message and contains the name of the format (protocol) that will be used for the duration of the connection. If the server does not support the protocol requested by the client or the first message received from the client is not a `Negotiation` message the server must close the connection.
+After opening a connection to the server the client must send a `HandshakeRequest` message to the server as its first message. The handshake message is **always** a JSON message and contains the name of the format (protocol) that will be used for the duration of the connection. The server will reply with a `HandshakeResponse`, also always JSON, containing an error if the server does not support the protocol. If the server does not support the protocol requested by the client or the first message received from the client is not a `HandshakeRequest` message the server must close the connection.
 
-The `Negotiation` message contains the following properties:
+The `HandshakeRequest` message contains the following properties:
 
 * `protocol` - the name of the protocol to be used for messages exchanged between the server and the client
 
@@ -42,6 +46,18 @@ Example:
 }
 ```
 
+The `HandshakeResponse` message contains the following properties:
+
+* `error` - the optional error message if the server does not support the request protocol
+
+Example:
+
+```json
+{
+    "error": "Requested protocol 'messagepack' is not available."
+}
+```
+
 ## Communication between the Caller and the Callee
 
 There are three kinds of interactions between the Caller and the Callee:
@@ -323,7 +339,7 @@ Example:
 
 A `StreamItem` message is a JSON object with the following properties:
 
-* `type` - A `Number` with the literal value 2, indicating that this message is a StreamItem.
+* `type` - A `Number` with the literal value 2, indicating that this message is a `StreamItem`.
 * `invocationId` - A `String` encoding the `Invocation ID` for a message.
 * `item` - A `Token` encoding the stream item (see "JSON Payload Encoding" for details).
 
@@ -391,7 +407,7 @@ Example - The following `Completion` message is a protocol error because it has
 ### CancelInvocation Message Encoding
 A `CancelInvocation` message is a JSON object with the following properties
 
-* `type` - A `Number` with the literal value `5`, indicationg that this is a `CancelInvocation`.
+* `type` - A `Number` with the literal value `5`, indicating that this message is a `CancelInvocation`.
 * `invocationId` - A `String` encoding the `Invocation ID` for a message.
 
 Example
@@ -405,7 +421,7 @@ Example
 ### Ping Message Encoding
 A `Ping` message is a JSON object with the following properties:
 
-* `type` - A `Number` with the literal value `6`, indicating that this is a `Ping`.
+* `type` - A `Number` with the literal value `6`, indicating that this message is a `Ping`.
 
 Example
 ```json
@@ -414,6 +430,27 @@ Example
 }
 ```
 
+### Close Message Encoding
+A `Close` message is a JSON object with the following properties
+
+* `type` - A `Number` with the literal value `7`, indicating that this message is a `Close`.
+* `error` - An optional `String` encoding the error message.
+
+Example - A `Close` message without an error
+```json
+{
+    "type": 7
+}
+```
+
+Example - A `Close` message with an error
+```json
+{
+    "type": 7,
+    "error": "Connection closed because of an error!"
+}
+```
+
 ### JSON Header Encoding
 
 Message headers are encoded into a JSON object, with string values, that are stored in the `headers` property. For example:
@@ -714,9 +751,38 @@ The following payload:
 
 is decoded as follows:
 
-* `0x92` - 2-element array
+* `0x91` - 1-element array
 * `0x06` - `6` (Message Type - `Ping` message)
 
+### Close Message Encoding
+
+`Close` messages have the following structure
+
+```
+[7, Error]
+```
+
+* `7` - Message Type - `7` indicates this is a `Close` message.
+* `Error` - Error - A `String` encoding the error for the message.
+
+Examples:
+
+#### Close message
+
+The following payload:
+```
+0x92 0x07 0xa3 0x78 0x79 0x7a
+```
+
+is decoded as follows:
+
+* `0x92` - 2-element array
+* `0x07` - `7` (Message Type - `Close` message)
+* `0xa3` - string of length 3 (Error)
+* `0x78` - `x`
+* `0x79` - `y`
+* `0x7a` - `z`
+
 ### MessagePack Headers Encoding
 
 Headers are encoded in MessagePack messages as a Map that immediately follows the type value. The Map can be empty, in which case it is represented by the byte `0x80`. If there are items in the map,

src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.Log.cs

@@ -92,8 +92,8 @@ namespace Microsoft.AspNetCore.SignalR.Client
             private static readonly Action<ILogger, Exception> _errorDuringClosedEvent =
                 LoggerMessage.Define(LogLevel.Error, new EventId(27, "ErrorDuringClosedEvent"), "An exception was thrown in the handler for the Closed event.");
 
-            private static readonly Action<ILogger, Exception> _sendingHubNegotiate =
+            private static readonly Action<ILogger, Exception> _sendingHubHandshake =
-                LoggerMessage.Define(LogLevel.Debug, new EventId(28, "SendingHubNegotiate"), "Sending Hub Negotiation.");
+                LoggerMessage.Define(LogLevel.Debug, new EventId(28, "SendingHubHandshake"), "Sending Hub Handshake.");
 
             private static readonly Action<ILogger, int, Exception> _parsingMessages =
                 LoggerMessage.Define<int>(LogLevel.Debug, new EventId(29, "ParsingMessages"), "Received {Count} bytes. Parsing message(s).");
@@ -113,6 +113,18 @@ namespace Microsoft.AspNetCore.SignalR.Client
             private static readonly Action<ILogger, string, Exception> _errorInvokingClientSideMethod =
                 LoggerMessage.Define<string>(LogLevel.Error, new EventId(34, "ErrorInvokingClientSideMethod"), "Invoking client side method '{MethodName}' failed.");
 
+            private static readonly Action<ILogger, Exception> _errorProcessingHandshakeResponse =
+                LoggerMessage.Define(LogLevel.Error, new EventId(35, "ErrorReceivingHandshakeResponse"), "Error processing the handshake response.");
+
+            private static readonly Action<ILogger, string, Exception> _handshakeServerError =
+                LoggerMessage.Define<string>(LogLevel.Error, new EventId(36, "HandshakeServerError"), "Server returned handshake error: {Error}");
+
+            private static readonly Action<ILogger, Exception> _receivedClose =
+                LoggerMessage.Define(LogLevel.Debug, new EventId(37, "ReceivedClose"), "Received close message.");
+
+            private static readonly Action<ILogger, string, Exception> _receivedCloseWithError =
+                LoggerMessage.Define<string>(LogLevel.Error, new EventId(38, "ReceivedCloseWithError"), "Received close message with an error: {Error}");
+
             public static void PreparingNonBlockingInvocation(ILogger logger, string target, int count)
             {
                 _preparingNonBlockingInvocation(logger, target, count, null);
@@ -256,9 +268,9 @@ namespace Microsoft.AspNetCore.SignalR.Client
                 _errorDuringClosedEvent(logger, exception);
             }
 
-            public static void SendingHubNegotiate(ILogger logger)
+            public static void SendingHubHandshake(ILogger logger)
             {
-                _sendingHubNegotiate(logger, null);
+                _sendingHubHandshake(logger, null);
             }
 
             public static void ParsingMessages(ILogger logger, int byteCount)
@@ -290,6 +302,26 @@ namespace Microsoft.AspNetCore.SignalR.Client
             {
                 _errorInvokingClientSideMethod(logger, methodName, exception);
             }
+
+            public static void ErrorReceivingHandshakeResponse(ILogger logger, Exception exception)
+            {
+                _errorProcessingHandshakeResponse(logger, exception);
+            }
+
+            public static void HandshakeServerError(ILogger logger, string error)
+            {
+                _handshakeServerError(logger, error, null);
+            }
+
+            public static void ReceivedClose(ILogger logger)
+            {
+                _receivedClose(logger, null);
+            }
+
+            public static void ReceivedCloseWithError(ILogger logger, string error)
+            {
+                _receivedCloseWithError(logger, error, null);
+            }
         }
     }
 }

src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.cs

@@ -10,6 +10,7 @@ using System.Threading.Channels;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Protocols.Features;
 using Microsoft.AspNetCore.SignalR.Internal;
+using Microsoft.AspNetCore.SignalR.Internal.Formatters;
 using Microsoft.AspNetCore.SignalR.Internal.Protocol;
 using Microsoft.AspNetCore.Sockets.Client;
 using Microsoft.AspNetCore.Sockets.Internal;
@@ -33,10 +34,11 @@ namespace Microsoft.AspNetCore.SignalR.Client
         private readonly ConcurrentDictionary<string, List<InvocationHandler>> _handlers = new ConcurrentDictionary<string, List<InvocationHandler>>();
         private CancellationTokenSource _connectionActive;
 
-        private int _nextId = 0;
+        private int _nextId;
         private volatile bool _startCalled;
-        private Timer _timeoutTimer;
+        private readonly Timer _timeoutTimer;
         private bool _needKeepAlive;
+        private bool _receivedHandshakeResponse;
 
         public event Action<Exception> Closed;
 
@@ -64,7 +66,7 @@ namespace Microsoft.AspNetCore.SignalR.Client
             _loggerFactory = loggerFactory ?? NullLoggerFactory.Instance;
             _logger = _loggerFactory.CreateLogger<HubConnection>();
             _connection.OnReceived((data, state) => ((HubConnection)state).OnDataReceivedAsync(data), this);
-            _connection.Closed += e => Shutdown(e);
+            _connection.Closed += Shutdown;
 
             // Create the timer for timeout, but disabled by default (we enable it when started).
             _timeoutTimer = new Timer(state => ((HubConnection)state).TimeoutElapsed(), this, Timeout.Infinite, Timeout.Infinite);
@@ -111,14 +113,15 @@ namespace Microsoft.AspNetCore.SignalR.Client
         {
             await _connection.StartAsync(_protocol.TransferFormat);
             _needKeepAlive = _connection.Features.Get<IConnectionInherentKeepAliveFeature>() == null;
+            _receivedHandshakeResponse = false;
 
             Log.HubProtocol(_logger, _protocol.Name);
 
             _connectionActive = new CancellationTokenSource();
             using (var memoryStream = new MemoryStream())
             {
-                Log.SendingHubNegotiate(_logger);
+                Log.SendingHubHandshake(_logger);
-                NegotiationProtocol.WriteMessage(new NegotiationMessage(_protocol.Name), memoryStream);
+                HandshakeProtocol.WriteRequestMessage(new HandshakeRequestMessage(_protocol.Name), memoryStream);
                 await _connection.SendAsync(memoryStream.ToArray(), _connectionActive.Token);
             }
 
@@ -309,9 +312,28 @@ namespace Microsoft.AspNetCore.SignalR.Client
         private async Task OnDataReceivedAsync(byte[] data)
         {
             ResetTimeoutTimer();
-            Log.ParsingMessages(_logger, data.Length);
+
+            var currentData = new ReadOnlyMemory<byte>(data);
+            Log.ParsingMessages(_logger, currentData.Length);
+
+            // first message received must be handshake response
+            if (!_receivedHandshakeResponse)
+            {
+                // process handshake and return left over data to parse additional messages
+                if (!ProcessHandshakeResponse(ref currentData))
+                {
+                    return;
+                }
+
+                _receivedHandshakeResponse = true;
+                if (currentData.IsEmpty)
+                {
+                    return;
+                }
+            }
+
             var messages = new List<HubMessage>();
-            if (_protocol.TryParseMessages(data, _binder, messages))
+            if (_protocol.TryParseMessages(currentData, _binder, messages))
             {
                 Log.ReceivingMessages(_logger, messages.Count);
                 foreach (var message in messages)
@@ -342,6 +364,18 @@ namespace Microsoft.AspNetCore.SignalR.Client
                             }
                             DispatchInvocationStreamItemAsync(streamItem, irq);
                             break;
+                        case CloseMessage close:
+                            if (string.IsNullOrEmpty(close.Error))
+                            {
+                                Log.ReceivedClose(_logger);
+                                Shutdown();
+                            }
+                            else
+                            {
+                                Log.ReceivedCloseWithError(_logger, close.Error);
+                                Shutdown(new InvalidOperationException(close.Error));
+                            }
+                            break;
                         case PingMessage _:
                             Log.ReceivedPing(_logger);
                             // Nothing to do on receipt of a ping.
@@ -358,8 +392,47 @@ namespace Microsoft.AspNetCore.SignalR.Client
             }
         }
 
+        private bool ProcessHandshakeResponse(ref ReadOnlyMemory<byte> data)
+        {
+            HandshakeResponseMessage message;
+
+            try
+            {
+                // read first message out of the incoming data
+                if (!TextMessageParser.TryParseMessage(ref data, out var payload))
+                {
+                    throw new InvalidDataException("Unable to parse payload as a handshake response message.");
+                }
+
+                message = HandshakeProtocol.ParseResponseMessage(payload);
+            }
+            catch (Exception ex)
+            {
+                // shutdown if we're unable to read handshake
+                Log.ErrorReceivingHandshakeResponse(_logger, ex);
+                Shutdown(ex);
+                return false;
+            }
+
+            if (!string.IsNullOrEmpty(message.Error))
+            {
+                // shutdown if handshake returns an error
+                Log.HandshakeServerError(_logger, message.Error);
+                Shutdown();
+                return false;
+            }
+
+            return true;
+        }
+
         private void Shutdown(Exception exception = null)
         {
+            // check if connection has already been shutdown
+            if (_connectionActive.IsCancellationRequested)
+            {
+                return;
+            }
+
             Log.ShutdownConnection(_logger);
             if (exception != null)
             {

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/CloseMessage.cs

@@ -0,0 +1,20 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
+{
+    public class CloseMessage : HubMessage
+    {
+        public static readonly CloseMessage Empty = new CloseMessage(null);
+
+        public string Error { get; }
+
+        public CloseMessage(string error)
+        {
+            Error = error;
+        }
+    }
+}

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HandshakeProtocol.cs

@@ -0,0 +1,128 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.Collections;
+using System.IO;
+using System.Text;
+using Microsoft.AspNetCore.SignalR.Internal.Formatters;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+
+namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
+{
+    public static class HandshakeProtocol
+    {
+        private static readonly UTF8Encoding _utf8NoBom = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false);
+
+        private const string ProtocolPropertyName = "protocol";
+        private const string ErrorPropertyName = "error";
+        private const string TypePropertyName = "type";
+
+        public static void WriteRequestMessage(HandshakeRequestMessage requestMessage, Stream output)
+        {
+            using (var writer = CreateJsonTextWriter(output))
+            {
+                writer.WriteStartObject();
+                writer.WritePropertyName(ProtocolPropertyName);
+                writer.WriteValue(requestMessage.Protocol);
+                writer.WriteEndObject();
+            }
+
+            TextMessageFormatter.WriteRecordSeparator(output);
+        }
+
+        public static void WriteResponseMessage(HandshakeResponseMessage responseMessage, Stream output)
+        {
+            using (var writer = CreateJsonTextWriter(output))
+            {
+                writer.WriteStartObject();
+                if (!string.IsNullOrEmpty(responseMessage.Error))
+                {
+                    writer.WritePropertyName(ErrorPropertyName);
+                    writer.WriteValue(responseMessage.Error);
+                }
+                writer.WriteEndObject();
+            }
+
+            TextMessageFormatter.WriteRecordSeparator(output);
+        }
+
+        private static JsonTextWriter CreateJsonTextWriter(Stream output)
+        {
+            return new JsonTextWriter(new StreamWriter(output, _utf8NoBom, 1024, leaveOpen: true));
+        }
+
+        private static JsonTextReader CreateJsonTextReader(ReadOnlyMemory<byte> payload)
+        {
+            var textReader = new Utf8BufferTextReader(payload);
+            var reader = new JsonTextReader(textReader);
+            reader.ArrayPool = JsonArrayPool<char>.Shared;
+
+            return reader;
+        }
+
+        public static HandshakeResponseMessage ParseResponseMessage(ReadOnlyMemory<byte> payload)
+        {
+            using (var reader = CreateJsonTextReader(payload))
+            {
+                var token = JToken.ReadFrom(reader);
+                var handshakeJObject = JsonUtils.GetObject(token);
+
+                // a handshake response does not have a type
+                // check the incoming message was not any other type of message
+                var type = JsonUtils.GetOptionalProperty<string>(handshakeJObject, TypePropertyName);
+                if (!string.IsNullOrEmpty(type))
+                {
+                    throw new InvalidOperationException("Handshake response should not have a 'type' value.");
+                }
+
+                var error = JsonUtils.GetOptionalProperty<string>(handshakeJObject, ErrorPropertyName);
+                return new HandshakeResponseMessage(error);
+            }
+        }
+
+        public static bool TryParseRequestMessage(ReadOnlySequence<byte> buffer, out HandshakeRequestMessage requestMessage, out SequencePosition consumed, out SequencePosition examined)
+        {
+            if (!TryReadMessageIntoSingleMemory(buffer, out consumed, out examined, out var memory))
+            {
+                requestMessage = null;
+                return false;
+            }
+
+            if (!TextMessageParser.TryParseMessage(ref memory, out var payload))
+            {
+                throw new InvalidDataException("Unable to parse payload as a handshake request message.");
+            }
+
+            using (var reader = CreateJsonTextReader(payload))
+            {
+                var token = JToken.ReadFrom(reader);
+                var handshakeJObject = JsonUtils.GetObject(token);
+                var protocol = JsonUtils.GetRequiredProperty<string>(handshakeJObject, ProtocolPropertyName);
+                requestMessage = new HandshakeRequestMessage(protocol);
+            }
+
+            return true;
+        }
+
+        internal static bool TryReadMessageIntoSingleMemory(ReadOnlySequence<byte> buffer, out SequencePosition consumed, out SequencePosition examined, out ReadOnlyMemory<byte> memory)
+        {
+            var separator = buffer.PositionOf(TextMessageFormatter.RecordSeparator);
+            if (separator == null)
+            {
+                // Haven't seen the entire message so bail
+                consumed = buffer.Start;
+                examined = buffer.End;
+                memory = null;
+                return false;
+            }
+
+            consumed = buffer.GetPosition(1, separator.Value);
+            examined = consumed;
+            memory = buffer.IsSingleSegment ? buffer.First : buffer.ToArray();
+            return true;
+        }
+    }
+}

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HandshakeRequestMessage.cs

@@ -3,9 +3,9 @@
 
 namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
 {
-    public class NegotiationMessage
+    public class HandshakeRequestMessage : HubMessage
     {
-        public NegotiationMessage(string protocol)
+        public HandshakeRequestMessage(string protocol)
         {
             Protocol = protocol;
         }

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HandshakeResponseMessage.cs

@@ -0,0 +1,17 @@
+// 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.
+
+namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
+{
+    public class HandshakeResponseMessage : HubMessage
+    {
+        public static readonly HandshakeResponseMessage Empty = new HandshakeResponseMessage(null);
+
+        public string Error { get; }
+
+        public HandshakeResponseMessage(string error)
+        {
+            Error = error;
+        }
+    }
+}

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HubProtocolConstants.cs

@@ -11,5 +11,6 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
         public const int StreamInvocationMessageType = 4;
         public const int CancelInvocationMessageType = 5;
         public const int PingMessageType = 6;
+        public const int CloseMessageType = 7;
     }
 }

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/JsonHubProtocol.cs

@@ -204,14 +204,12 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
                             case JsonToken.EndObject:
                                 completed = true;
                                 break;
-                            default:
-                                break;
                         }
                     }
                     while (!completed && JsonUtils.CheckRead(reader));
                 }
 
-                HubMessage message = null;
+                HubMessage message;
 
                 switch (type)
                 {
@@ -274,6 +272,8 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
                         break;
                     case HubProtocolConstants.PingMessageType:
                         return PingMessage.Instance;
+                    case HubProtocolConstants.CloseMessageType:
+                        return BindCloseMessage(error);
                     case null:
                         throw new InvalidDataException($"Missing required property '{TypePropertyName}'.");
                     default:
@@ -358,6 +358,10 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
                     case PingMessage _:
                         WriteMessageType(writer, HubProtocolConstants.PingMessageType);
                         break;
+                    case CloseMessage m:
+                        WriteMessageType(writer, HubProtocolConstants.CloseMessageType);
+                        WriteCloseMessage(m, writer);
+                        break;
                     default:
                         throw new InvalidOperationException($"Unsupported message type: {message.GetType().FullName}");
                 }
@@ -425,6 +429,15 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
             WriteArguments(message.Arguments, writer);
         }
 
+        private void WriteCloseMessage(CloseMessage message, JsonTextWriter writer)
+        {
+            if (!string.IsNullOrEmpty(message.Error))
+            {
+                writer.WritePropertyName(ErrorPropertyName);
+                writer.WriteValue(message.Error);
+            }
+        }
+
         private void WriteArguments(object[] arguments, JsonTextWriter writer)
         {
             writer.WritePropertyName(ArgumentsPropertyName);
@@ -569,6 +582,17 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
             throw new JsonReaderException("Unexpected end when reading JSON");
         }
 
+        private CloseMessage BindCloseMessage(string error)
+        {
+            if (string.IsNullOrEmpty(error))
+            {
+                return CloseMessage.Empty;
+            }
+
+            var message = new CloseMessage(error);
+            return message;
+        }
+
         private object[] BindArguments(JArray args, IReadOnlyList<Type> paramTypes)
         {
             var arguments = new object[args.Count];

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/JsonUtils.cs

@@ -10,6 +10,16 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
 {
     public static class JsonUtils
     {
+        public static JObject GetObject(JToken token)
+        {
+            if (token == null || token.Type != JTokenType.Object)
+            {
+                throw new InvalidDataException($"Unexpected JSON Token Type '{token?.Type}'. Expected a JSON Object.");
+            }
+
+            return (JObject)token;
+        }
+
         public static T GetOptionalProperty<T>(JObject json, string property, JTokenType expectedType = JTokenType.None, T defaultValue = default)
         {
             var prop = json[property];

src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/NegotiationProtocol.cs

@@ -1,80 +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.
-
-using System;
-using System.Buffers;
-using System.IO;
-using System.Text;
-using Microsoft.AspNetCore.SignalR.Internal.Formatters;
-using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
-
-namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
-{
-    public static class NegotiationProtocol
-    {
-        private static readonly UTF8Encoding _utf8NoBom = new UTF8Encoding(encoderShouldEmitUTF8Identifier: false);
-
-        private const string ProtocolPropertyName = "protocol";
-
-        public static void WriteMessage(NegotiationMessage negotiationMessage, Stream output)
-        {
-            using (var writer = new JsonTextWriter(new StreamWriter(output, _utf8NoBom, 1024, leaveOpen: true)))
-            {
-                writer.WriteStartObject();
-                writer.WritePropertyName(ProtocolPropertyName);
-                writer.WriteValue(negotiationMessage.Protocol);
-                writer.WriteEndObject();
-            }
-
-            TextMessageFormatter.WriteRecordSeparator(output);
-        }
-
-        public static bool TryParseMessage(ReadOnlyMemory<byte> input, out NegotiationMessage negotiationMessage)
-        {
-            if (!TextMessageParser.TryParseMessage(ref input, out var payload))
-            {
-                throw new InvalidDataException("Unable to parse payload as a negotiation message.");
-            }
-
-            var textReader = new Utf8BufferTextReader(payload);
-            using (var reader = new JsonTextReader(textReader))
-            {
-                reader.ArrayPool = JsonArrayPool<char>.Shared;
-
-                var token = JToken.ReadFrom(reader);
-                if (token == null || token.Type != JTokenType.Object)
-                {
-                    throw new InvalidDataException($"Unexpected JSON Token Type '{token?.Type}'. Expected a JSON Object.");
-                }
-
-                var negotiationJObject = (JObject)token;
-                var protocol = JsonUtils.GetRequiredProperty<string>(negotiationJObject, ProtocolPropertyName);
-                negotiationMessage = new NegotiationMessage(protocol);
-            }
-            return true;
-        }
-
-        public static bool TryParseMessage(ReadOnlySequence<byte> buffer, out NegotiationMessage negotiationMessage, out SequencePosition consumed, out SequencePosition examined)
-        {
-            var separator = buffer.PositionOf(TextMessageFormatter.RecordSeparator);
-            if (separator == null)
-            {
-                // Haven't seen the entire negotiate message so bail
-                consumed = buffer.Start;
-                examined = buffer.End;
-                negotiationMessage = null;
-                return false;
-            }
-            else
-            {
-                consumed = buffer.GetPosition(1, separator.Value);
-                examined = consumed;
-            }
-
-            var memory = buffer.IsSingleSegment ? buffer.First : buffer.ToArray();
-
-            return TryParseMessage(memory, out negotiationMessage);
-        }
-    }
-}

src/Microsoft.AspNetCore.SignalR.Common/Properties/AssemblyInfo.cs

@@ -0,0 +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.
+
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Microsoft.AspNetCore.SignalR.Tests.Utils, PublicKey=0024000004800000940000000602000000240000525341310004000001000100f33a29044fa9d740c9b3213a93e57c84b472c84e0b8a0e1ae48e67a9f8f6de9d5f7f3d52ac23e48ac51801f1dc950abe901da34d2a9e3baadb141a17c77ef3c565dd5ee5054b91cf63bb3c6ab83f72ab3aafe93d0fc3c2348b764fafb0b1c0733de51459aeab46580384bf9d74c4e28164b7cde247f891ba07891c9d872ad2bb")]

src/Microsoft.AspNetCore.SignalR.Core/HubConnectionContext.cs

@@ -6,6 +6,7 @@ using System.Buffers;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.IO;
 using System.IO.Pipelines;
 using System.Net;
 using System.Runtime.ExceptionServices;
@@ -118,6 +119,23 @@ namespace Microsoft.AspNetCore.SignalR
             }
         }
 
+        private async Task WriteHandshakeResponseAsync(HandshakeResponseMessage message)
+        {
+            await _writeLock.WaitAsync();
+
+            try
+            {
+                var ms = new MemoryStream();
+                HandshakeProtocol.WriteResponseMessage(message, ms);
+
+                await _connectionContext.Transport.Output.WriteAsync(ms.ToArray());
+            }
+            finally
+            {
+                _writeLock.Release();
+            }
+        }
+
         public virtual void Abort()
         {
             // If we already triggered the token then noop, this isn't thread safe but it's good enough
@@ -131,7 +149,7 @@ namespace Microsoft.AspNetCore.SignalR
             Task.Factory.StartNew(_abortedCallback, this);
         }
 
-        internal async Task<bool> NegotiateAsync(TimeSpan timeout, IList<string> supportedProtocols, IHubProtocolResolver protocolResolver, IUserIdProvider userIdProvider)
+        internal async Task<bool> HandshakeAsync(TimeSpan timeout, IList<string> supportedProtocols, IHubProtocolResolver protocolResolver, IUserIdProvider userIdProvider)
         {
             try
             {
@@ -150,9 +168,16 @@ namespace Microsoft.AspNetCore.SignalR
                         {
                             if (!buffer.IsEmpty)
                             {
-                                if (NegotiationProtocol.TryParseMessage(buffer, out var negotiationMessage, out consumed, out examined))
+                                if (HandshakeProtocol.TryParseRequestMessage(buffer, out var handshakeRequestMessage, out consumed, out examined))
                                 {
-                                    Protocol = protocolResolver.GetProtocol(negotiationMessage.Protocol, supportedProtocols, this);
+                                    Protocol = protocolResolver.GetProtocol(handshakeRequestMessage.Protocol, supportedProtocols, this);
+                                    if (Protocol == null)
+                                    {
+                                        Log.HandshakeFailed(_logger, null);
+
+                                        await WriteHandshakeResponseAsync(new HandshakeResponseMessage($"The protocol '{handshakeRequestMessage.Protocol}' is not supported."));
+                                        return false;
+                                    }
 
                                     // If there's a transfer format feature, we need to check if we're compatible and set the active format.
                                     // If there isn't a feature, it means that the transport supports binary data and doesn't need us to tell them
@@ -162,7 +187,9 @@ namespace Microsoft.AspNetCore.SignalR
                                     {
                                         if ((transferFormatFeature.SupportedFormats & Protocol.TransferFormat) == 0)
                                         {
-                                            throw new InvalidOperationException($"Cannot use the '{Protocol.Name}' protocol on the current transport. The transport does not support the '{Protocol.TransferFormat}' transfer mode.");
+                                            Log.HandshakeFailed(_logger, null);
+                                            await WriteHandshakeResponseAsync(new HandshakeResponseMessage($"Cannot use the '{Protocol.Name}' protocol on the current transport. The transport does not support '{Protocol.TransferFormat}' transfer format."));
+                                            return false;
                                         }
 
                                         transferFormatFeature.ActiveFormat = Protocol.TransferFormat;
@@ -170,22 +197,25 @@ namespace Microsoft.AspNetCore.SignalR
 
                                     _cachedPingMessage = Protocol.WriteToArray(PingMessage.Instance);
 
-                                    Log.UsingHubProtocol(_logger, Protocol.Name);
-
                                     UserIdentifier = userIdProvider.GetUserId(this);
 
                                     if (Features.Get<IConnectionInherentKeepAliveFeature>() == null)
                                     {
-                                        // Only register KeepAlive after protocol negotiated otherwise KeepAliveTick could try to write without having a ProtocolReaderWriter
+                                        // Only register KeepAlive after protocol handshake otherwise KeepAliveTick could try to write without having a ProtocolReaderWriter
                                         Features.Get<IConnectionHeartbeatFeature>()?.OnHeartbeat(state => ((HubConnectionContext)state).KeepAliveTick(), this);
                                     }
 
+                                    Log.HandshakeComplete(_logger, Protocol.Name);
+                                    await WriteHandshakeResponseAsync(HandshakeResponseMessage.Empty);
                                     return true;
                                 }
                             }
                             else if (result.IsCompleted)
                             {
-                                break;
+                                // connection was closed before we ever received a response
+                                // can't send a handshake response because there is no longer a connection
+                                Log.HandshakeFailed(_logger, null);
+                                return false;
                             }
                         }
                         finally
@@ -197,10 +227,16 @@ namespace Microsoft.AspNetCore.SignalR
             }
             catch (OperationCanceledException)
             {
-                Log.NegotiateCanceled(_logger);
+                Log.HandshakeCanceled(_logger);
+                await WriteHandshakeResponseAsync(new HandshakeResponseMessage("Handshake was canceled."));
+                return false;
+            }
+            catch (Exception ex)
+            {
+                Log.HandshakeFailed(_logger, ex);
+                await WriteHandshakeResponseAsync(new HandshakeResponseMessage($"An unexpected error occurred during connection handshake. {ex.GetType().Name}: {ex.Message}"));
+                return false;
             }
-
-            return false;
         }
 
         internal void Abort(Exception exception)
@@ -257,11 +293,11 @@ namespace Microsoft.AspNetCore.SignalR
         private static class Log
         {
             // Category: HubConnectionContext
-            private static readonly Action<ILogger, string, Exception> _usingHubProtocol =
+            private static readonly Action<ILogger, string, Exception> _handshakeComplete =
-                LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, "UsingHubProtocol"), "Using HubProtocol '{Protocol}'.");
+                LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, "HandshakeComplete"), "Completed connection handshake. Using HubProtocol '{Protocol}'.");
 
-            private static readonly Action<ILogger, Exception> _negotiateCanceled =
+            private static readonly Action<ILogger, Exception> _handshakeCanceled =
-                LoggerMessage.Define(LogLevel.Debug, new EventId(2, "NegotiateCanceled"), "Negotiate was canceled.");
+                LoggerMessage.Define(LogLevel.Debug, new EventId(2, "HandshakeCanceled"), "Handshake was canceled.");
 
             private static readonly Action<ILogger, Exception> _sentPing =
                 LoggerMessage.Define(LogLevel.Trace, new EventId(3, "SentPing"), "Sent a ping message to the client.");
@@ -269,14 +305,17 @@ namespace Microsoft.AspNetCore.SignalR
             private static readonly Action<ILogger, Exception> _transportBufferFull =
                 LoggerMessage.Define(LogLevel.Debug, new EventId(4, "TransportBufferFull"), "Unable to send Ping message to client, the transport buffer is full.");
 
-            public static void UsingHubProtocol(ILogger logger, string hubProtocol)
+            private static readonly Action<ILogger, Exception> _handshakeFailed =
+                LoggerMessage.Define(LogLevel.Error, new EventId(5, "HandshakeFailed"), "Failed connection handshake.");
+
+            public static void HandshakeComplete(ILogger logger, string hubProtocol)
             {
-                _usingHubProtocol(logger, hubProtocol, null);
+                _handshakeComplete(logger, hubProtocol, null);
             }
 
-            public static void NegotiateCanceled(ILogger logger)
+            public static void HandshakeCanceled(ILogger logger)
             {
-                _negotiateCanceled(logger, null);
+                _handshakeCanceled(logger, null);
             }
 
             public static void SentPing(ILogger logger)
@@ -288,6 +327,11 @@ namespace Microsoft.AspNetCore.SignalR
             {
                 _transportBufferFull(logger, null);
             }
+
+            public static void HandshakeFailed(ILogger logger, Exception exception)
+            {
+                _handshakeFailed(logger, exception);
+            }
         }
 
     }

src/Microsoft.AspNetCore.SignalR.Core/HubEndPoint.cs

@@ -46,9 +46,9 @@ namespace Microsoft.AspNetCore.SignalR
         public async Task OnConnectedAsync(ConnectionContext connection)
         {
             // We check to see if HubOptions<THub> are set because those take precedence over global hub options.
-            // Then set the keepAlive and negotiateTimeout values to the defaults in HubOptionsSetup incase they were explicitly set to null.
+            // Then set the keepAlive and handshakeTimeout values to the defaults in HubOptionsSetup incase they were explicitly set to null.
             var keepAlive = _hubOptions.KeepAliveInterval ?? _globalHubOptions.KeepAliveInterval ?? HubOptionsSetup.DefaultKeepAliveInterval;
-            var negotiateTimeout = _hubOptions.NegotiateTimeout ?? _globalHubOptions.NegotiateTimeout ?? HubOptionsSetup.DefaultNegotiateTimeout;
+            var handshakeTimeout = _hubOptions.HandshakeTimeout ?? _globalHubOptions.HandshakeTimeout ?? HubOptionsSetup.DefaultHandshakeTimeout;
             var supportedProtocols = _hubOptions.SupportedProtocols ?? _globalHubOptions.SupportedProtocols;
 
             if (supportedProtocols != null && supportedProtocols.Count == 0)
@@ -58,7 +58,7 @@ namespace Microsoft.AspNetCore.SignalR
 
             var connectionContext = new HubConnectionContext(connection, keepAlive, _loggerFactory);
 
-            if (!await connectionContext.NegotiateAsync(negotiateTimeout, supportedProtocols, _protocolResolver, _userIdProvider))
+            if (!await connectionContext.HandshakeAsync(handshakeTimeout, supportedProtocols, _protocolResolver, _userIdProvider))
             {
                 return;
             }
@@ -83,7 +83,11 @@ namespace Microsoft.AspNetCore.SignalR
             catch (Exception ex)
             {
                 Log.ErrorDispatchingHubEvent(_logger, "OnConnectedAsync", ex);
-                throw;
+
+                await SendCloseAsync(connection, ex);
+
+                // return instead of throw to let close message send successfully
+                return;
             }
 
             try
@@ -93,8 +97,11 @@ namespace Microsoft.AspNetCore.SignalR
             catch (Exception ex)
             {
                 Log.ErrorProcessingRequest(_logger, ex);
+
                 await HubOnDisconnectedAsync(connection, ex);
-                throw;
+
+                // return instead of throw to let close message send successfully
+                return;
             }
 
             await HubOnDisconnectedAsync(connection, null);
@@ -102,6 +109,9 @@ namespace Microsoft.AspNetCore.SignalR
 
         private async Task HubOnDisconnectedAsync(HubConnectionContext connection, Exception exception)
         {
+            // send close message before aborting the connection
+            await SendCloseAsync(connection, exception);
+
             // We wait on abort to complete, this is so that we can guarantee that all callbacks have fired
             // before OnDisconnectedAsync
 
@@ -126,6 +136,22 @@ namespace Microsoft.AspNetCore.SignalR
             }
         }
 
+        private async Task SendCloseAsync(HubConnectionContext connection, Exception exception)
+        {
+            CloseMessage closeMessage = exception == null
+                ? CloseMessage.Empty
+                : new CloseMessage($"Connection closed with an error. {exception.GetType().Name}: {exception.Message}");
+
+            try
+            {
+                await connection.WriteAsync(closeMessage);
+            }
+            catch (Exception ex)
+            {
+                Log.ErrorSendingClose(_logger, ex);
+            }
+        }
+
         private async Task DispatchMessagesAsync(HubConnectionContext connection)
         {
             // Since we dispatch multiple hub invocations in parallel, we need a way to communicate failure back to the main processing loop.
@@ -186,6 +212,9 @@ namespace Microsoft.AspNetCore.SignalR
             private static readonly Action<ILogger, Exception> _abortFailed =
                 LoggerMessage.Define(LogLevel.Trace, new EventId(3, "AbortFailed"), "Abort callback failed.");
 
+            private static readonly Action<ILogger, Exception> _errorSendingClose =
+                LoggerMessage.Define(LogLevel.Debug, new EventId(4, "ErrorSendingClose"), "Error when sending Close message.");
+
             public static void ErrorDispatchingHubEvent(ILogger logger, string hubMethod, Exception exception)
             {
                 _errorDispatchingHubEvent(logger, hubMethod, exception);
@@ -200,6 +229,11 @@ namespace Microsoft.AspNetCore.SignalR
             {
                 _abortFailed(logger, exception);
             }
+
+            public static void ErrorSendingClose(ILogger logger, Exception exception)
+            {
+                _errorSendingClose(logger, exception);
+            }
         }
     }
 }

src/Microsoft.AspNetCore.SignalR.Core/HubOptions.cs

@@ -8,11 +8,11 @@ namespace Microsoft.AspNetCore.SignalR
 {
     public class HubOptions
     {
-        // NegotiateTimeout and KeepAliveInterval are set to null here to help identify when
+        // HandshakeTimeout and KeepAliveInterval are set to null here to help identify when
         // local hub options have been set. Global default values are set in HubOptionsSetup.
         // SupportedProtocols being null is the true default value, and it represents support
         // for all available protocols.
-        public TimeSpan? NegotiateTimeout { get; set; } = null;
+        public TimeSpan? HandshakeTimeout { get; set; } = null;
 
         public TimeSpan? KeepAliveInterval { get; set; } = null;
 

src/Microsoft.AspNetCore.SignalR.Core/HubOptionsSetup.cs

@@ -11,7 +11,7 @@ namespace Microsoft.AspNetCore.SignalR
 {
     public class HubOptionsSetup : IConfigureOptions<HubOptions>
     {
-        internal static TimeSpan DefaultNegotiateTimeout => TimeSpan.FromSeconds(5);
+        internal static TimeSpan DefaultHandshakeTimeout => TimeSpan.FromSeconds(5);
 
         internal static TimeSpan DefaultKeepAliveInterval => TimeSpan.FromSeconds(15);
 
@@ -39,9 +39,9 @@ namespace Microsoft.AspNetCore.SignalR
                 options.KeepAliveInterval = DefaultKeepAliveInterval;
             }
 
-            if (options.NegotiateTimeout == null)
+            if (options.HandshakeTimeout == null)
             {
-                options.NegotiateTimeout = DefaultNegotiateTimeout;
+                options.HandshakeTimeout = DefaultHandshakeTimeout;
             }
 
             foreach (var protocol in _protocols)

src/Microsoft.AspNetCore.SignalR.Core/HubOptionsSetup`T.cs

@@ -18,7 +18,7 @@ namespace Microsoft.AspNetCore.SignalR
         {
             options.SupportedProtocols = _hubOptions.SupportedProtocols;
             options.KeepAliveInterval = _hubOptions.KeepAliveInterval;
-            options.NegotiateTimeout = _hubOptions.NegotiateTimeout;
+            options.HandshakeTimeout = _hubOptions.HandshakeTimeout;
         }
     }
 }

src/Microsoft.AspNetCore.SignalR.Core/Internal/DefaultHubProtocolResolver.cs

@@ -42,7 +42,9 @@ namespace Microsoft.AspNetCore.SignalR.Internal
                 return protocol;
             }
 
-            throw new NotSupportedException($"The protocol '{protocolName ?? "(null)"}' is not supported.");
+            // null result indicates protocol is not supported
+            // result will be validated by the caller
+            return null;
         }
 
         private static class Log

src/Microsoft.AspNetCore.SignalR.Protocols.MsgPack/Internal/Protocol/MessagePackHubProtocol.cs

@@ -69,6 +69,8 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
                         return CreateCancelInvocationMessage(unpacker);
                     case HubProtocolConstants.PingMessageType:
                         return PingMessage.Instance;
+                    case HubProtocolConstants.CloseMessageType:
+                        return CreateCloseMessage(unpacker);
                     default:
                         throw new FormatException($"Invalid message type: {messageType}.");
                 }
@@ -165,6 +167,12 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
             return ApplyHeaders(headers, new CancelInvocationMessage(invocationId));
         }
 
+        private static CloseMessage CreateCloseMessage(Unpacker unpacker)
+        {
+            var error = ReadString(unpacker, "error");
+            return new CloseMessage(error);
+        }
+
         private static Dictionary<string, string> ReadHeaders(Unpacker unpacker)
         {
             var headerCount = ReadMapLength(unpacker, "headers");
@@ -269,6 +277,9 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
                 case PingMessage pingMessage:
                     WritePingMessage(pingMessage, packer);
                     break;
+                case CloseMessage closeMessage:
+                    WriteCloseMessage(closeMessage, packer);
+                    break;
                 default:
                     throw new FormatException($"Unexpected message type: {message.GetType().Name}");
             }
@@ -341,6 +352,20 @@ namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
             packer.PackString(message.InvocationId);
         }
 
+        private void WriteCloseMessage(CloseMessage message, Packer packer)
+        {
+            packer.PackArrayHeader(2);
+            packer.Pack(HubProtocolConstants.CloseMessageType);
+            if (string.IsNullOrEmpty(message.Error))
+            {
+                packer.PackNull();
+            }
+            else
+            {
+                packer.PackString(message.Error);
+            }
+        }
+
         private void WritePingMessage(PingMessage pingMessage, Packer packer)
         {
             packer.PackArrayHeader(1);

src/Microsoft.AspNetCore.Sockets.Abstractions/ConnectionMetadata.cs

@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 
 namespace Microsoft.AspNetCore.Sockets
@@ -10,7 +11,7 @@ namespace Microsoft.AspNetCore.Sockets
     internal class ConnectionMetadata : IDictionary<object, object>
     {
         public ConnectionMetadata()
-            : this(new Dictionary<object, object>())
+            : this(new ConcurrentDictionary<object, object>())
         {
         }
 

src/Microsoft.AspNetCore.Sockets.Http/HttpConnectionDispatcher.cs

@@ -33,7 +33,7 @@ namespace Microsoft.AspNetCore.Sockets
             _logger = _loggerFactory.CreateLogger<HttpConnectionDispatcher>();
         }
 
-        public async Task ExecuteAsync(HttpContext context, HttpSocketOptions options, ConnectionDelegate ConnectionDelegate)
+        public async Task ExecuteAsync(HttpContext context, HttpSocketOptions options, ConnectionDelegate connectionDelegate)
         {
             // Create the log scope and attempt to pass the Connection ID to it so as many logs as possible contain
             // the Connection ID metadata. If this is the negotiate request then the Connection ID for the scope will
@@ -54,7 +54,7 @@ namespace Microsoft.AspNetCore.Sockets
                 else if (HttpMethods.IsGet(context.Request.Method))
                 {
                     // GET /{path}
-                    await ExecuteEndpointAsync(context, ConnectionDelegate, options, logScope);
+                    await ExecuteEndpointAsync(context, connectionDelegate, options, logScope);
                 }
                 else
                 {
@@ -88,7 +88,7 @@ namespace Microsoft.AspNetCore.Sockets
             }
         }
 
-        private async Task ExecuteEndpointAsync(HttpContext context, ConnectionDelegate ConnectionDelegate, HttpSocketOptions options, ConnectionLogScope logScope)
+        private async Task ExecuteEndpointAsync(HttpContext context, ConnectionDelegate connectionDelegate, HttpSocketOptions options, ConnectionLogScope logScope)
         {
             var supportedTransports = options.Transports;
 
@@ -120,7 +120,7 @@ namespace Microsoft.AspNetCore.Sockets
                 // We only need to provide the Input channel since writing to the application is handled through /send.
                 var sse = new ServerSentEventsTransport(connection.Application.Input, connection.ConnectionId, _loggerFactory);
 
-                await DoPersistentConnection(ConnectionDelegate, sse, context, connection);
+                await DoPersistentConnection(connectionDelegate, sse, context, connection);
             }
             else if (context.WebSockets.IsWebSocketRequest)
             {
@@ -142,7 +142,7 @@ namespace Microsoft.AspNetCore.Sockets
 
                 var ws = new WebSocketsTransport(options.WebSockets, connection.Application, connection, _loggerFactory);
 
-                await DoPersistentConnection(ConnectionDelegate, ws, context, connection);
+                await DoPersistentConnection(connectionDelegate, ws, context, connection);
             }
             else
             {
@@ -203,7 +203,7 @@ namespace Microsoft.AspNetCore.Sockets
 
                         connection.Items[ConnectionMetadataNames.Transport] = TransportType.LongPolling;
 
-                        connection.ApplicationTask = ExecuteApplication(ConnectionDelegate, connection);
+                        connection.ApplicationTask = ExecuteApplication(connectionDelegate, connection);
                     }
                     else
                     {
@@ -292,7 +292,7 @@ namespace Microsoft.AspNetCore.Sockets
             }
         }
 
-        private async Task DoPersistentConnection(ConnectionDelegate ConnectionDelegate,
+        private async Task DoPersistentConnection(ConnectionDelegate connectionDelegate,
                                                   IHttpTransport transport,
                                                   HttpContext context,
                                                   DefaultConnectionContext connection)
@@ -324,7 +324,7 @@ namespace Microsoft.AspNetCore.Sockets
                 connection.Status = DefaultConnectionContext.ConnectionStatus.Active;
 
                 // Call into the end point passing the connection
-                connection.ApplicationTask = ExecuteApplication(ConnectionDelegate, connection);
+                connection.ApplicationTask = ExecuteApplication(connectionDelegate, connection);
 
                 // Start the transport
                 connection.TransportTask = transport.ProcessRequestAsync(context, context.RequestAborted);
@@ -340,7 +340,7 @@ namespace Microsoft.AspNetCore.Sockets
             await _manager.DisposeAndRemoveAsync(connection);
         }
 
-        private async Task ExecuteApplication(ConnectionDelegate ConnectionDelegate, ConnectionContext connection)
+        private async Task ExecuteApplication(ConnectionDelegate connectionDelegate, ConnectionContext connection)
         {
             // Verify some initialization invariants
             // We want to be positive that the IConnectionInherentKeepAliveFeature is initialized before invoking the application, if the long polling transport is in use.
@@ -353,7 +353,7 @@ namespace Microsoft.AspNetCore.Sockets
             await AwaitableThreadPool.Yield();
 
             // Running this in an async method turns sync exceptions into async ones
-            await ConnectionDelegate(connection);
+            await connectionDelegate(connection);
         }
 
         private Task ProcessNegotiate(HttpContext context, HttpSocketOptions options, ConnectionLogScope logScope)

test/Common/ChannelExtensions.cs

@@ -1,25 +0,0 @@
-using System.Collections.Generic;
-using System.Threading.Tasks;
-
-namespace System.Threading.Channels
-{
-    internal static class ChannelExtensions
-    {
-        public static async Task<List<T>> ReadAllAsync<T>(this ChannelReader<T> channel)
-        {
-            var list = new List<T>();
-            while (await channel.WaitToReadAsync())
-            {
-                while (channel.TryRead(out var item))
-                {
-                    list.Add(item);
-                }
-            }
-
-            // Manifest any error from channel.Completion (which should be completed now)
-            await channel.Completion;
-
-            return list;
-        }
-    }
-}

test/Common/ServerFixture.cs

@@ -1,112 +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.
-
-using System;
-using System.IO;
-using System.Net;
-using System.Net.Sockets;
-using System.Threading.Tasks;
-using Microsoft.AspNetCore.Hosting;
-using Microsoft.Extensions.DependencyInjection;
-using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Logging.Testing;
-
-
-namespace Microsoft.AspNetCore.SignalR.Tests.Common
-{
-    public class ServerFixture<TStartup> : IDisposable
-        where TStartup : class
-    {
-        private readonly ILoggerFactory _loggerFactory;
-        private readonly ILogger _logger;
-        private IWebHost _host;
-        private IApplicationLifetime _lifetime;
-        private readonly IDisposable _logToken;
-
-        public string WebSocketsUrl => Url.Replace("http", "ws");
-
-        public string Url { get; private set; }
-
-        public ServerFixture()
-        {
-            var testLog = AssemblyTestLog.ForAssembly(typeof(ServerFixture<TStartup>).Assembly);
-            _logToken = testLog.StartTestLog(null, $"{nameof(ServerFixture<TStartup>)}_{typeof(TStartup).Name}", out _loggerFactory, "ServerFixture");
-            _logger = _loggerFactory.CreateLogger<ServerFixture<TStartup>>();
-            Url = "http://localhost:" + GetNextPort();
-
-            StartServer(Url);
-        }
-
-        private void StartServer(string url)
-        {
-            _host = new WebHostBuilder()
-                .ConfigureLogging(builder => builder.AddProvider(new ForwardingLoggerProvider(_loggerFactory)))
-                .UseStartup(typeof(TStartup))
-                .UseKestrel()
-                .UseUrls(url)
-                .UseContentRoot(Directory.GetCurrentDirectory())
-                .Build();
-
-            var t = Task.Run(() => _host.Start());
-            _logger.LogInformation("Starting test server...");
-            _lifetime = _host.Services.GetRequiredService<IApplicationLifetime>();
-            if (!_lifetime.ApplicationStarted.WaitHandle.WaitOne(TimeSpan.FromSeconds(5)))
-            {
-                // t probably faulted
-                if (t.IsFaulted)
-                {
-                    throw t.Exception.InnerException;
-                }
-                throw new TimeoutException("Timed out waiting for application to start.");
-            }
-            _logger.LogInformation("Test Server started");
-
-            _lifetime.ApplicationStopped.Register(() =>
-            {
-                _logger.LogInformation("Test server shut down");
-                _logToken.Dispose();
-            });
-        }
-
-        public void Dispose()
-        {
-            _logger.LogInformation("Shutting down test server");
-            _host.Dispose();
-            _loggerFactory.Dispose();
-        }
-
-        private class ForwardingLoggerProvider : ILoggerProvider
-        {
-            private readonly ILoggerFactory _loggerFactory;
-
-            public ForwardingLoggerProvider(ILoggerFactory loggerFactory)
-            {
-                _loggerFactory = loggerFactory;
-            }
-
-            public void Dispose()
-            {
-            }
-
-            public ILogger CreateLogger(string categoryName)
-            {
-                return _loggerFactory.CreateLogger(categoryName);
-            }
-        }
-
-        // Copied from https://github.com/aspnet/KestrelHttpServer/blob/47f1db20e063c2da75d9d89653fad4eafe24446c/test/Microsoft.AspNetCore.Server.Kestrel.FunctionalTests/AddressRegistrationTests.cs#L508
-        private static int GetNextPort()
-        {
-            using (var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp))
-            {
-                // Let the OS assign the next available port. Unless we cycle through all ports
-                // on a test run, the OS will always increment the port number when making these calls.
-                // This prevents races in parallel test runs where a test is already bound to
-                // a given port, and a new test is able to bind to the same port due to port
-                // reuse being enabled by default by the OS.
-                socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
-                return ((IPEndPoint)socket.LocalEndPoint).Port;
-            }
-        }
-    }
-}

test/Common/TestClient.cs

@@ -1,213 +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.
-
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.Security.Claims;
-using System.Threading;
-using System.Threading.Tasks;
-using System.Threading.Channels;
-using Microsoft.AspNetCore.SignalR.Internal;
-using Microsoft.AspNetCore.SignalR.Internal.Encoders;
-using Microsoft.AspNetCore.SignalR.Internal.Protocol;
-using Microsoft.AspNetCore.Sockets;
-using Microsoft.AspNetCore.Sockets.Internal;
-using Newtonsoft.Json;
-
-namespace Microsoft.AspNetCore.SignalR.Tests
-{
-    public class TestClient : IDisposable
-    {
-        private static int _id;
-        private readonly HubProtocolReaderWriter _protocolReaderWriter;
-        private readonly IInvocationBinder _invocationBinder;
-        private CancellationTokenSource _cts;
-        private ChannelConnection<byte[]> _transport;
-
-        public DefaultConnectionContext Connection { get; }
-        public Channel<byte[]> Application { get; }
-        public Task Connected => ((TaskCompletionSource<bool>)Connection.Metadata["ConnectedTask"]).Task;
-
-        public TestClient(bool synchronousCallbacks = false, IHubProtocol protocol = null, IInvocationBinder invocationBinder = null, bool addClaimId = false)
-        {
-            var options = new UnboundedChannelOptions { AllowSynchronousContinuations = synchronousCallbacks };
-            var transportToApplication = Channel.CreateUnbounded<byte[]>(options);
-            var applicationToTransport = Channel.CreateUnbounded<byte[]>(options);
-
-            Application = ChannelConnection.Create<byte[]>(input: applicationToTransport, output: transportToApplication);
-            _transport = ChannelConnection.Create<byte[]>(input: transportToApplication, output: applicationToTransport);
-
-            Connection = new DefaultConnectionContext(Guid.NewGuid().ToString(), _transport, Application);
-
-            var claimValue = Interlocked.Increment(ref _id).ToString();
-            var claims = new List<Claim>{ new Claim(ClaimTypes.Name, claimValue) };
-            if (addClaimId)
-            {
-                claims.Add(new Claim(ClaimTypes.NameIdentifier, claimValue));
-            }
-
-            Connection.User = new ClaimsPrincipal(new ClaimsIdentity(claims));
-            Connection.Metadata["ConnectedTask"] = new TaskCompletionSource<bool>();
-
-            protocol = protocol ?? new JsonHubProtocol();
-            _protocolReaderWriter = new HubProtocolReaderWriter(protocol, new PassThroughEncoder());
-            _invocationBinder = invocationBinder ?? new DefaultInvocationBinder();
-
-            _cts = new CancellationTokenSource();
-
-            using (var memoryStream = new MemoryStream())
-            {
-                NegotiationProtocol.WriteMessage(new NegotiationMessage(protocol.Name), memoryStream);
-                Application.Writer.TryWrite(memoryStream.ToArray());
-            }
-        }
-
-        public async Task<IList<HubMessage>> StreamAsync(string methodName, params object[] args)
-        {
-            var invocationId = await SendStreamInvocationAsync(methodName, args);
-
-            var messages = new List<HubMessage>();
-            while (true)
-            {
-                var message = await ReadAsync();
-
-                if (message == null)
-                {
-                    throw new InvalidOperationException("Connection aborted!");
-                }
-
-                if (message is HubInvocationMessage hubInvocationMessage && !string.Equals(hubInvocationMessage.InvocationId, invocationId))
-                {
-                    throw new NotSupportedException("TestClient does not support multiple outgoing invocations!");
-                }
-
-                switch (message)
-                {
-                    case StreamItemMessage _:
-                        messages.Add(message);
-                        break;
-                    case CompletionMessage _:
-                        messages.Add(message);
-                        return messages;
-                    default:
-                        throw new NotSupportedException("TestClient does not support receiving invocations!");
-                }
-            }
-        }
-
-        public async Task<CompletionMessage> InvokeAsync(string methodName, params object[] args)
-        {
-            var invocationId = await SendInvocationAsync(methodName, nonBlocking: false, args: args);
-
-            while (true)
-            {
-                var message = await ReadAsync();
-
-                if (message == null)
-                {
-                    throw new InvalidOperationException("Connection aborted!");
-                }
-
-                if (message is HubInvocationMessage hubInvocationMessage && !string.Equals(hubInvocationMessage.InvocationId, invocationId))
-                {
-                    throw new NotSupportedException("TestClient does not support multiple outgoing invocations!");
-                }
-
-                switch (message)
-                {
-                    case StreamItemMessage result:
-                        throw new NotSupportedException("Use 'StreamAsync' to call a streaming method");
-                    case CompletionMessage completion:
-                        return completion;
-                    case PingMessage _:
-                        // Pings are ignored
-                        break;
-                    default:
-                        throw new NotSupportedException("TestClient does not support receiving invocations!");
-                }
-            }
-        }
-
-        public Task<string> SendInvocationAsync(string methodName, params object[] args)
-        {
-            return SendInvocationAsync(methodName, nonBlocking: false, args: args);
-        }
-
-        public Task<string> SendInvocationAsync(string methodName, bool nonBlocking, params object[] args)
-        {
-            var invocationId = GetInvocationId();
-            return SendHubMessageAsync(new InvocationMessage(invocationId, nonBlocking, methodName,
-                argumentBindingException: null, arguments: args));
-        }
-
-        public Task<string> SendStreamInvocationAsync(string methodName, params object[] args)
-        {
-            var invocationId = GetInvocationId();
-            return SendHubMessageAsync(new StreamInvocationMessage(invocationId, methodName,
-                argumentBindingException: null, arguments: args));
-        }
-
-        public async Task<string> SendHubMessageAsync(HubMessage message)
-        {
-            var payload = _protocolReaderWriter.WriteMessage(message);
-            await Application.Writer.WriteAsync(payload);
-            return message is HubInvocationMessage hubMessage ? hubMessage.InvocationId : null;
-        }
-
-        public async Task<HubMessage> ReadAsync()
-        {
-            while (true)
-            {
-                var message = TryRead();
-
-                if (message == null)
-                {
-                    if (!await Application.Reader.WaitToReadAsync())
-                    {
-                        return null;
-                    }
-                }
-                else
-                {
-                    return message;
-                }
-            }
-        }
-
-        public HubMessage TryRead()
-        {
-            if (Application.Reader.TryRead(out var buffer) &&
-                _protocolReaderWriter.ReadMessages(buffer, _invocationBinder, out var messages))
-            {
-                return messages[0];
-            }
-            return null;
-        }
-
-        public void Dispose()
-        {
-            _cts.Cancel();
-            _transport.Dispose();
-        }
-
-        private static string GetInvocationId()
-        {
-            return Guid.NewGuid().ToString("N");
-        }
-
-        private class DefaultInvocationBinder : IInvocationBinder
-        {
-            public Type[] GetParameterTypes(string methodName)
-            {
-                // TODO: Possibly support actual client methods
-                return new[] { typeof(object) };
-            }
-
-            public Type GetReturnType(string invocationId)
-            {
-                return typeof(object);
-            }
-        }
-    }
-}

test/Common/TestHelpers.cs

@@ -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.
-
-using System;
-
-namespace Microsoft.AspNetCore.SignalR.Tests.Common
-{
-	public static class TestHelpers
-    {
-        public static bool IsWebSocketsSupported()
-        {
-            try
-            {
-                new System.Net.WebSockets.ClientWebSocket().Dispose();
-            }
-            catch (PlatformNotSupportedException)
-            {
-                return false;
-            }
-
-            return true;
-        }
-    }
-}

test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs

@@ -107,7 +107,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
         [MemberData(nameof(HubProtocolsAndTransportsAndHubPaths))]
         public async Task CanStopAndStartConnection(IHubProtocol protocol, TransportType transportType, string path)
         {
-            using (StartLog(out var loggerFactory))
+            using (StartLog(out var loggerFactory, LogLevel.Trace, $"{nameof(CanStopAndStartConnection)}_{protocol.Name}_{transportType}_{path.TrimStart('/')}"))
             {
                 const string originalMessage = "SignalR";
                 var httpConnection = new HttpConnection(new Uri(_serverFixture.Url + path), transportType, loggerFactory);
@@ -692,7 +692,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
             }
         }
 
-        [Theory]
+        [Theory(Skip = "HttpContext + Long Polling fails. Issue logged - https://github.com/aspnet/SignalR/issues/1644")]
         [MemberData(nameof(TransportTypes))]
         public async Task ClientCanSendHeaders(TransportType transportType)
         {

test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/Hubs.cs

@@ -37,7 +37,20 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests
 
         public IEnumerable<string> GetHeaderValues(string[] headerNames)
         {
-            var headers = Context.Connection.GetHttpContext().Request.Headers;
+            var context = Context.Connection.GetHttpContext();
+
+            if (context == null)
+            {
+                throw new InvalidOperationException("Unable to get HttpContext from request.");
+            }
+
+            var headers = context.Request.Headers;
+
+            if (headers == null)
+            {
+                throw new InvalidOperationException("Unable to get headers from context.");
+            }
+
             return headerNames.Select(h => (string)headers[h]);
         }
 

test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionExtensionsTests.cs

@@ -102,6 +102,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 onAction(hubConnection, handlerTcs);
                 await hubConnection.StartAsync();
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
 
                 await connection.ReceiveJsonMessage(
                     new
@@ -133,6 +134,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
                 hubConnection.On<int>("Foo", r => { receiveTcs.SetResult(r); });
                 await hubConnection.StartAsync().OrTimeout();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 await connection.ReceiveJsonMessage(
                     new
                     {
@@ -170,6 +173,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 hubConnection.On<int>("Foo", r => { receiveTcs.SetResult(r); });
                 await hubConnection.StartAsync().OrTimeout();
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
 
                 await connection.ReceiveJsonMessage(
                     new

test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionProtocolTests.cs

@@ -29,10 +29,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var invokeTask = hubConnection.SendAsync("Foo");
 
-                // skip negotiation
-                await connection.ReadSentTextMessageAsync().OrTimeout();
                 var invokeMessage = await connection.ReadSentTextMessageAsync().OrTimeout();
 
                 Assert.Equal("{\"type\":1,\"target\":\"Foo\",\"arguments\":[]}\u001e", invokeMessage);
@@ -45,16 +45,17 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
         }
 
         [Fact]
-        public async Task ClientSendsNegotationMessageWhenStartingConnection()
+        public async Task ClientSendsHandshakeMessageWhenStartingConnection()
         {
             var connection = new TestConnection();
             var hubConnection = new HubConnection(connection, new JsonHubProtocol(), new LoggerFactory());
             try
             {
                 await hubConnection.StartAsync();
-                var negotiationMessage = await connection.ReadSentTextMessageAsync().OrTimeout();
 
-                Assert.Equal("{\"protocol\":\"json\"}\u001e", negotiationMessage);
+                var handshakeMessage = await connection.ReadSentTextMessageAsync().OrTimeout();
+
+                Assert.Equal("{\"protocol\":\"json\"}\u001e", handshakeMessage);
             }
             finally
             {
@@ -72,10 +73,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var invokeTask = hubConnection.InvokeAsync("Foo");
 
-                // skip negotiation
-                await connection.ReadSentTextMessageAsync().OrTimeout();
                 var invokeMessage = await connection.ReadSentTextMessageAsync().OrTimeout();
 
                 Assert.Equal("{\"type\":1,\"invocationId\":\"1\",\"target\":\"Foo\",\"arguments\":[]}\u001e", invokeMessage);
@@ -87,6 +88,61 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             }
         }
 
+        [Fact]
+        public async Task ReceiveCloseMessageWithoutErrorWillCloseHubConnection()
+        {
+            TaskCompletionSource<Exception> closedTcs = new TaskCompletionSource<Exception>();
+
+            var connection = new TestConnection();
+            var hubConnection = new HubConnection(connection, new JsonHubProtocol(), new LoggerFactory());
+            hubConnection.Closed += e => closedTcs.SetResult(e);
+
+            try
+            {
+                await hubConnection.StartAsync();
+
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
+                await connection.ReceiveJsonMessage(new { type = 7 }).OrTimeout();
+
+                Exception closeException = await closedTcs.Task.OrTimeout();
+                Assert.Null(closeException);
+            }
+            finally
+            {
+                await hubConnection.DisposeAsync().OrTimeout();
+                await connection.DisposeAsync().OrTimeout();
+            }
+        }
+
+        [Fact]
+        public async Task ReceiveCloseMessageWithErrorWillCloseHubConnection()
+        {
+            TaskCompletionSource<Exception> closedTcs = new TaskCompletionSource<Exception>();
+
+            var connection = new TestConnection();
+            var hubConnection = new HubConnection(connection, new JsonHubProtocol(), new LoggerFactory());
+            hubConnection.Closed += e => closedTcs.SetResult(e);
+
+            try
+            {
+                await hubConnection.StartAsync();
+
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
+                await connection.ReceiveJsonMessage(new { type = 7, error = "Error!" }).OrTimeout();
+
+                Exception closeException = await closedTcs.Task.OrTimeout();
+                Assert.NotNull(closeException);
+                Assert.Equal("Error!", closeException.Message);
+            }
+            finally
+            {
+                await hubConnection.DisposeAsync().OrTimeout();
+                await connection.DisposeAsync().OrTimeout();
+            }
+        }
+
         [Fact]
         public async Task StreamSendsAnInvocationMessage()
         {
@@ -96,10 +152,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var channel = await hubConnection.StreamAsChannelAsync<object>("Foo");
 
-                // skip negotiation
-                await connection.ReadSentTextMessageAsync().OrTimeout();
                 var invokeMessage = await connection.ReadSentTextMessageAsync().OrTimeout();
 
                 Assert.Equal("{\"type\":4,\"invocationId\":\"1\",\"target\":\"Foo\",\"arguments\":[]}\u001e", invokeMessage);
@@ -124,6 +180,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var invokeTask = hubConnection.InvokeAsync("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 3 }).OrTimeout();
@@ -146,6 +204,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var channel = await hubConnection.StreamAsChannelAsync<int>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 3 }).OrTimeout();
@@ -168,6 +228,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var invokeTask = hubConnection.InvokeAsync<int>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 3, result = 42 }).OrTimeout();
@@ -190,6 +252,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var invokeTask = hubConnection.InvokeAsync<int>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 3, error = "An error occurred" }).OrTimeout();
@@ -213,6 +277,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var channel = await hubConnection.StreamAsChannelAsync<string>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 3, result = "Oops" }).OrTimeout();
@@ -236,6 +302,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var channel = await hubConnection.StreamAsChannelAsync<int>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 3, error = "An error occurred" }).OrTimeout();
@@ -259,6 +327,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var invokeTask = hubConnection.InvokeAsync<int>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 2, item = 42 }).OrTimeout();
@@ -282,6 +352,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 var channel = await hubConnection.StreamAsChannelAsync<string>("Foo");
 
                 await connection.ReceiveJsonMessage(new { invocationId = "1", type = 2, item = "1" }).OrTimeout();
@@ -310,6 +382,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync();
 
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
+
                 hubConnection.On<int, string, float>("Foo", (r1, r2, r3) => handlerCalled.TrySetResult(new object[] { r1, r2, r3 }));
 
                 var args = new object[] { 1, "Foo", 2.0f };
@@ -335,8 +409,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             {
                 await hubConnection.StartAsync().OrTimeout();
 
-                // Ignore negotiate message
+                // Ignore handshake message
-                await connection.ReadSentTextMessageAsync().OrTimeout();
+                await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
 
                 // Send an invocation
                 var invokeTask = hubConnection.InvokeAsync("Foo");

test/Microsoft.AspNetCore.SignalR.Client.Tests/TestConnection.cs

@@ -11,6 +11,7 @@ using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http.Features;
 using Microsoft.AspNetCore.Protocols;
 using Microsoft.AspNetCore.SignalR.Internal.Formatters;
+using Microsoft.AspNetCore.SignalR.Internal.Protocol;
 using Microsoft.AspNetCore.Sockets.Client;
 using Newtonsoft.Json;
 
@@ -81,6 +82,16 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
             return Task.CompletedTask;
         }
 
+        public async Task ReadHandshakeAndSendResponseAsync()
+        {
+            await SentMessages.ReadAsync();
+
+            var output = new MemoryStream();
+            HandshakeProtocol.WriteResponseMessage(HandshakeResponseMessage.Empty, output);
+
+            await _receivedMessages.Writer.WriteAsync(output.ToArray());
+        }
+
         public async Task<string> ReadSentTextMessageAsync()
         {
             var message = await SentMessages.ReadAsync();

test/Microsoft.AspNetCore.SignalR.Common.Tests/Internal/Protocol/HandshakeProtocolTests.cs

@@ -0,0 +1,81 @@
+// 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.
+
+using System;
+using System.Buffers;
+using System.IO;
+using System.Text;
+using Microsoft.AspNetCore.SignalR.Internal.Protocol;
+using Xunit;
+
+namespace Microsoft.AspNetCore.SignalR.Common.Tests.Internal.Protocol
+{
+    public class HandshakeProtocolTests
+    {
+        [Theory]
+        [InlineData("{\"protocol\":\"dummy\"}\u001e", "dummy")]
+        [InlineData("{\"protocol\":\"\"}\u001e", "")]
+        [InlineData("{\"protocol\":null}\u001e", null)]
+        public void ParsingHandshakeRequestMessageSuccessForValidMessages(string json, string protocol)
+        {
+            var message = Encoding.UTF8.GetBytes(json);
+
+            Assert.True(HandshakeProtocol.TryParseRequestMessage(new ReadOnlySequence<byte>(message), out var deserializedMessage, out _, out _));
+
+            Assert.Equal(protocol, deserializedMessage.Protocol);
+        }
+
+        [Theory]
+        [InlineData("{\"error\":\"dummy\"}\u001e", "dummy")]
+        [InlineData("{\"error\":\"\"}\u001e", "")]
+        [InlineData("{\"error\":null}\u001e", null)]
+        [InlineData("{}\u001e", null)]
+        public void ParsingHandshakeResponseMessageSuccessForValidMessages(string json, string error)
+        {
+            var message = Encoding.UTF8.GetBytes(json);
+
+            var response = HandshakeProtocol.ParseResponseMessage(message);
+
+            Assert.Equal(error, response.Error);
+        }
+
+        [Fact]
+        public void ParsingHandshakeRequestNotCompleteReturnsFalse()
+        {
+            var message = Encoding.UTF8.GetBytes("42");
+
+            Assert.False(HandshakeProtocol.TryParseRequestMessage(new ReadOnlySequence<byte>(message), out _, out _, out _));
+        }
+
+        [Theory]
+        [InlineData("42\u001e", "Unexpected JSON Token Type 'Integer'. Expected a JSON Object.")]
+        [InlineData("\"42\"\u001e", "Unexpected JSON Token Type 'String'. Expected a JSON Object.")]
+        [InlineData("null\u001e", "Unexpected JSON Token Type 'Null'. Expected a JSON Object.")]
+        [InlineData("{}\u001e", "Missing required property 'protocol'.")]
+        [InlineData("[]\u001e", "Unexpected JSON Token Type 'Array'. Expected a JSON Object.")]
+        public void ParsingHandshakeRequestMessageThrowsForInvalidMessages(string payload, string expectedMessage)
+        {
+            var message = Encoding.UTF8.GetBytes(payload);
+
+            var exception = Assert.Throws<InvalidDataException>(() =>
+                Assert.True(HandshakeProtocol.TryParseRequestMessage(new ReadOnlySequence<byte>(message), out _, out _, out _)));
+
+            Assert.Equal(expectedMessage, exception.Message);
+        }
+
+        [Theory]
+        [InlineData("42", "Unexpected JSON Token Type 'Integer'. Expected a JSON Object.")]
+        [InlineData("\"42\"", "Unexpected JSON Token Type 'String'. Expected a JSON Object.")]
+        [InlineData("null", "Unexpected JSON Token Type 'Null'. Expected a JSON Object.")]
+        [InlineData("[]", "Unexpected JSON Token Type 'Array'. Expected a JSON Object.")]
+        public void ParsingHandshakeResponseMessageThrowsForInvalidMessages(string payload, string expectedMessage)
+        {
+            var message = Encoding.UTF8.GetBytes(payload);
+
+            var exception = Assert.Throws<InvalidDataException>(() =>
+                HandshakeProtocol.ParseResponseMessage(message));
+
+            Assert.Equal(expectedMessage, exception.Message);
+        }
+    }
+}

test/Microsoft.AspNetCore.SignalR.Common.Tests/Internal/Protocol/NegotiationProtocolTests.cs

@@ -1,45 +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.
-
-using System;
-using System.IO;
-using System.Text;
-using Microsoft.AspNetCore.SignalR.Internal.Protocol;
-using Xunit;
-
-namespace Microsoft.AspNetCore.SignalR.Common.Tests.Internal.Protocol
-{
-    public class NegotiationProtocolTests
-    {
-        [Fact]
-        public void CanRoundtripNegotiation()
-        {
-            var negotiationMessage = new NegotiationMessage(protocol: "dummy");
-            using (var ms = new MemoryStream())
-            {
-                NegotiationProtocol.WriteMessage(negotiationMessage, ms);
-                Assert.True(NegotiationProtocol.TryParseMessage(ms.ToArray(), out var deserializedMessage));
-
-                Assert.NotNull(deserializedMessage);
-                Assert.Equal(negotiationMessage.Protocol, deserializedMessage.Protocol);
-            }
-        }
-
-        [Theory]
-        [InlineData("", "Unable to parse payload as a negotiation message.")]
-        [InlineData("42\u001e", "Unexpected JSON Token Type 'Integer'. Expected a JSON Object.")]
-        [InlineData("\"42\"\u001e", "Unexpected JSON Token Type 'String'. Expected a JSON Object.")]
-        [InlineData("null\u001e", "Unexpected JSON Token Type 'Null'. Expected a JSON Object.")]
-        [InlineData("{}\u001e", "Missing required property 'protocol'.")]
-        [InlineData("[]\u001e", "Unexpected JSON Token Type 'Array'. Expected a JSON Object.")]
-        public void ParsingNegotiationMessageThrowsForInvalidMessages(string payload, string expectedMessage)
-        {
-            var message = Encoding.UTF8.GetBytes(payload);
-
-            var exception = Assert.Throws<InvalidDataException>(() =>
-                Assert.True(NegotiationProtocol.TryParseMessage(message, out var deserializedMessage)));
-
-            Assert.Equal(expectedMessage, exception.Message);
-        }
-    }
-}

test/Microsoft.AspNetCore.SignalR.Tests.Utils/Microsoft.AspNetCore.SignalR.Tests.Utils.csproj

@@ -4,7 +4,11 @@
     <TargetFrameworks>$(StandardTestTfms)</TargetFrameworks>
     <RootNamespace>Microsoft.AspNetCore.SignalR.Tests</RootNamespace>
   </PropertyGroup>
-  
+
+  <ItemGroup Condition="'$(TargetFramework)' == 'net461'">
+    <PackageReference Include="Microsoft.CSharp" Version="$(MicrosoftCSharpPackageVersion)" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\..\src\Microsoft.AspNetCore.SignalR.Common\Microsoft.AspNetCore.SignalR.Common.csproj" />
     <ProjectReference Include="..\..\src\Microsoft.AspNetCore.SignalR.Core\Microsoft.AspNetCore.SignalR.Core.csproj" />

test/Microsoft.AspNetCore.SignalR.Tests.Utils/TestClient.cs

@@ -10,6 +10,7 @@ using System.Security.Claims;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.SignalR.Internal;
+using Microsoft.AspNetCore.SignalR.Internal.Formatters;
 using Microsoft.AspNetCore.SignalR.Internal.Protocol;
 using Microsoft.AspNetCore.Sockets;
 
@@ -20,11 +21,12 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         private static int _id;
         private readonly IHubProtocol _protocol;
         private readonly IInvocationBinder _invocationBinder;
-        private CancellationTokenSource _cts;
+        private readonly CancellationTokenSource _cts;
-        private Queue<HubMessage> _messages = new Queue<HubMessage>();
+        private readonly Queue<HubMessage> _messages = new Queue<HubMessage>();
 
         public DefaultConnectionContext Connection { get; }
         public Task Connected => ((TaskCompletionSource<bool>)Connection.Items["ConnectedTask"]).Task;
+        public HandshakeResponseMessage HandshakeResponseMessage { get; private set; }
 
         public TestClient(bool synchronousCallbacks = false, IHubProtocol protocol = null, IInvocationBinder invocationBinder = null, bool addClaimId = false)
         {
@@ -46,12 +48,33 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             _invocationBinder = invocationBinder ?? new DefaultInvocationBinder();
 
             _cts = new CancellationTokenSource();
+        }
 
-            using (var memoryStream = new MemoryStream())
+        public async Task<Task> ConnectAsync(
+            dynamic endPoint,
+            bool sendHandshakeRequestMessage = true,
+            bool expectedHandshakeResponseMessage = true)
+        {
+            if (sendHandshakeRequestMessage)
             {
-                NegotiationProtocol.WriteMessage(new NegotiationMessage(_protocol.Name), memoryStream);
+                using (var memoryStream = new MemoryStream())
-                Connection.Application.Output.WriteAsync(memoryStream.ToArray());
+                {
+                    HandshakeProtocol.WriteRequestMessage(new HandshakeRequestMessage(_protocol.Name), memoryStream);
+                    await Connection.Application.Output.WriteAsync(memoryStream.ToArray());
+                }
             }
+
+            var connection = (Task)endPoint.OnConnectedAsync(Connection);
+
+            if (expectedHandshakeResponseMessage)
+            {
+                // note that the handshake response might not immediately be readable
+                // e.g. server is waiting for request, times out after configured duration,
+                // and sends response with timeout error
+                HandshakeResponseMessage = (HandshakeResponseMessage) await ReadAsync(true).OrTimeout();
+            }
+
+            return connection;
         }
 
         public async Task<IList<HubMessage>> StreamAsync(string methodName, params object[] args)
@@ -147,11 +170,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             return message is HubInvocationMessage hubMessage ? hubMessage.InvocationId : null;
         }
 
-        public async Task<HubMessage> ReadAsync()
+        public async Task<HubMessage> ReadAsync(bool isHandshake = false)
         {
             while (true)
             {
-                var message = TryRead();
+                var message = TryRead(isHandshake);
 
                 if (message == null)
                 {
@@ -182,7 +205,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             }
         }
 
-        public HubMessage TryRead()
+        public HubMessage TryRead(bool isHandshake = false)
         {
             if (_messages.Count > 0)
             {
@@ -200,15 +223,30 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             try
             {
-                var messages = new List<HubMessage>();
+                if (!isHandshake)
-                if (_protocol.TryParseMessages(result.Buffer.ToArray(), _invocationBinder, messages))
                 {
-                    foreach (var m in messages)
+                    var messages = new List<HubMessage>();
+                    if (_protocol.TryParseMessages(result.Buffer.ToArray(), _invocationBinder, messages))
                     {
-                        _messages.Enqueue(m);
+                        foreach (var m in messages)
+                        {
+                            _messages.Enqueue(m);
+                        }
+
+                        return _messages.Dequeue();
+                    }
+                }
+                else
+                {
+                    HandshakeProtocol.TryReadMessageIntoSingleMemory(buffer, out consumed, out examined, out var data);
+
+                    // read first message out of the incoming data
+                    if (!TextMessageParser.TryParseMessage(ref data, out var payload))
+                    {
+                        throw new InvalidDataException("Unable to parse payload as a handshake response message.");
                     }
 
-                    return _messages.Dequeue();
+                    return HandshakeProtocol.ParseResponseMessage(payload);
                 }
             }
             finally

test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs

@@ -283,8 +283,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                     logger.LogInformation("Sent message", bytes.Length);
 
                     logger.LogInformation("Receiving message");
-                    // No timeout here because it can take a while to receive all the bytes
+                    // Big timeout here because it can take a while to receive all the bytes
-                    var receivedData = await receiveTcs.Task;
+                    var receivedData = await receiveTcs.Task.OrTimeout(TimeSpan.FromSeconds(30));
                     Assert.Equal(message, Encoding.UTF8.GetString(receivedData));
                     logger.LogInformation("Completed receive");
                 }
@@ -306,17 +306,29 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         [OSSkipCondition(OperatingSystems.Windows, WindowsVersions.Win7, WindowsVersions.Win2008R2, SkipReason = "No WebSockets Client for this platform")]
         public async Task ServerClosesConnectionWithErrorIfHubCannotBeCreated_WebSocket()
         {
-            var exception = await Assert.ThrowsAsync<InvalidOperationException>(
+            try
-                async () => await ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType.WebSockets));
+            {
-            Assert.Equal("Websocket closed with error: InternalServerError.", exception.Message);
+                await ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType.WebSockets);
+                Assert.True(false, "Expected error was not thrown.");
+            }
+            catch
+            {
+                // error is expected
+            }
         }
 
         [Fact]
         public async Task ServerClosesConnectionWithErrorIfHubCannotBeCreated_LongPolling()
         {
-            var exception = await Assert.ThrowsAsync<HttpRequestException>(
+            try
-                async () => await ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType.LongPolling));
+            {
-            Assert.Equal("Response status code does not indicate success: 500 (Internal Server Error).", exception.Message);
+                await ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType.LongPolling);
+                Assert.True(false, "Expected error was not thrown.");
+            }
+            catch
+            {
+                // error is expected
+            }
         }
 
         private async Task ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType transportType)
@@ -355,7 +367,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                     catch (OperationCanceledException)
                     {
                         // Due to a race, this can fail with OperationCanceledException in the SendAsync
-                        // call that HubConnection does to send the negotiate message.
+                        // call that HubConnection does to send the handshake message.
                         // This has only been happening on AppVeyor, likely due to a slower CI machine
                         // The closed event will still fire with the exception we care about.
                     }

test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTestUtils/Hubs.cs

@@ -405,7 +405,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests.HubEndpointTestUtils
 
     public class DisposeTrackingHub : TestHub
     {
-        private TrackDispose _trackDispose;
+        private readonly TrackDispose _trackDispose;
 
         public DisposeTrackingHub(TrackDispose trackDispose)
         {

test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTests.cs

@@ -2,9 +2,12 @@
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
 using System;
+using System.Buffers;
 using System.Collections.Generic;
+using System.IO;
 using System.Linq;
 using System.Security.Claims;
+using System.Text;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Protocols;
@@ -35,7 +38,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 // kill the connection
                 client.Dispose();
@@ -55,7 +58,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 // kill the connection
                 client.Dispose();
@@ -76,7 +79,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.InvokeAsync(nameof(AbortHub.Kill));
 
@@ -105,7 +108,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 async Task Produce()
                 {
@@ -168,7 +171,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 async Task Subscribe()
                 {
@@ -227,7 +230,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var invocationId = await client.SendStreamInvocationAsync(nameof(ObservableHub.Subscribe)).OrTimeout();
 
@@ -248,45 +251,44 @@ namespace Microsoft.AspNetCore.SignalR.Tests
         }
 
         [Fact]
-        public async Task MissingNegotiateAndMessageSentFromHubConnectionCanBeDisposedCleanly()
+        public async Task MissingHandshakeAndMessageSentFromHubConnectionCanBeDisposedCleanly()
         {
             var serviceProvider = HubEndPointTestUtils.CreateServiceProvider();
             var endPoint = serviceProvider.GetService<HubEndPoint<SimpleHub>>();
 
             using (var client = new TestClient())
             {
-                // TestClient automatically writes negotiate, for this test we want to assume negotiate never gets sent
+                Task endPointTask = await client.ConnectAsync(endPoint, false, false);
-                client.Connection.Transport.Input.TryRead(out var item);
-                client.Connection.Transport.Input.AdvanceTo(item.Buffer.End);
-
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
 
                 // kill the connection
                 client.Dispose();
 
                 await endPointTask;
+
+                Assert.Null(client.HandshakeResponseMessage);
             }
         }
 
         [Fact]
-        public async Task NegotiateTimesOut()
+        public async Task HandshakeTimesOut()
         {
             var serviceProvider = HubEndPointTestUtils.CreateServiceProvider(services =>
             {
                 services.Configure<HubOptions>(options =>
                 {
-                    options.NegotiateTimeout = TimeSpan.FromMilliseconds(5);
+                    options.HandshakeTimeout = TimeSpan.FromMilliseconds(5);
                 });
             });
             var endPoint = serviceProvider.GetService<HubEndPoint<SimpleHub>>();
 
             using (var client = new TestClient())
             {
-                // TestClient automatically writes negotiate, for this test we want to assume negotiate never gets sent
+                Task endPointTask = await client.ConnectAsync(endPoint, false);
-                client.Connection.Transport.Input.TryRead(out var item);
-                client.Connection.Transport.Input.AdvanceTo(item.Buffer.End);
 
-                await endPoint.OnConnectedAsync(client.Connection).OrTimeout();
+                Assert.NotNull(client.HandshakeResponseMessage);
+                Assert.Equal("Handshake was canceled.", client.HandshakeResponseMessage.Error);
+
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -306,6 +308,68 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             await context.Clients.All.Send("test");
         }
 
+        [Fact]
+        public async Task HandshakeFailureFromUnknownProtocolSendsResponseWithError()
+        {
+            var hubProtocolMock = new Mock<IHubProtocol>();
+            hubProtocolMock.Setup(m => m.Name).Returns("CustomProtocol");
+
+            dynamic endPoint = HubEndPointTestUtils.GetHubEndpoint(typeof(HubT));
+
+            using (var client = new TestClient(protocol: hubProtocolMock.Object))
+            {
+                Task endPointTask = await client.ConnectAsync(endPoint);
+
+                Assert.NotNull(client.HandshakeResponseMessage);
+                Assert.Equal("The protocol 'CustomProtocol' is not supported.", client.HandshakeResponseMessage.Error);
+
+                client.Dispose();
+
+                await endPointTask.OrTimeout();
+            }
+        }
+
+        [Fact]
+        public async Task HandshakeFailureFromUnsupportedFormatSendsResponseWithError()
+        {
+            var hubProtocolMock = new Mock<IHubProtocol>();
+            hubProtocolMock.Setup(m => m.Name).Returns("CustomProtocol");
+
+            dynamic endPoint = HubEndPointTestUtils.GetHubEndpoint(typeof(HubT));
+
+            using (var client = new TestClient(protocol: new MessagePackHubProtocol()))
+            {
+                client.Connection.SupportedFormats = TransferFormat.Text;
+
+                Task endPointTask = await client.ConnectAsync(endPoint);
+
+                Assert.NotNull(client.HandshakeResponseMessage);
+                Assert.Equal("Cannot use the 'messagepack' protocol on the current transport. The transport does not support 'Binary' transfer format.", client.HandshakeResponseMessage.Error);
+
+                client.Dispose();
+
+                await endPointTask.OrTimeout();
+            }
+        }
+
+        [Fact]
+        public async Task HandshakeSuccessSendsResponseWithoutError()
+        {
+            dynamic endPoint = HubEndPointTestUtils.GetHubEndpoint(typeof(HubT));
+
+            using (var client = new TestClient())
+            {
+                Task endPointTask = await client.ConnectAsync(endPoint);
+
+                Assert.NotNull(client.HandshakeResponseMessage);
+                Assert.Null(client.HandshakeResponseMessage.Error);
+
+                client.Dispose();
+
+                await endPointTask.OrTimeout();
+            }
+        }
+
         [Fact]
         public async Task LifetimeManagerOnDisconnectedAsyncCalledIfLifetimeManagerOnConnectedAsyncThrows()
         {
@@ -327,7 +391,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             {
                 var exception =
                     await Assert.ThrowsAsync<InvalidOperationException>(
-                        async () => await endPoint.OnConnectedAsync(client.Connection));
+                        async () =>
+                        {
+                            Task endPointTask = await client.ConnectAsync(endPoint);
+                            await endPointTask.OrTimeout();
+                        });
                 Assert.Equal("Lifetime manager OnConnectedAsync failed.", exception.Message);
 
                 client.Dispose();
@@ -353,11 +421,10 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
                 client.Dispose();
 
-                var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () => await endPointTask);
+                await endPointTask.OrTimeout();
-                Assert.Equal("Hub OnConnected failed.", exception.Message);
 
                 mockLifetimeManager.Verify(m => m.OnConnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
                 mockLifetimeManager.Verify(m => m.OnDisconnectedAsync(It.IsAny<HubConnectionContext>()), Times.Once);
@@ -377,7 +444,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
                 client.Dispose();
 
                 var exception = await Assert.ThrowsAsync<InvalidOperationException>(async () => await endPointTask);
@@ -397,7 +464,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync(nameof(MethodHub.TaskValueMethod)).OrTimeout()).Result;
 
@@ -419,7 +486,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                Task endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = (Task)await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync("echo", "hello").OrTimeout()).Result;
 
@@ -443,11 +510,11 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
-                var result = (await client.InvokeAsync(methodName).OrTimeout());
+                var message = await client.InvokeAsync(methodName).OrTimeout();
 
-                Assert.Equal($"An unexpected error occurred invoking '{methodName}' on the server. InvalidOperationException: BOOM!", result.Error);
+                Assert.Equal($"An unexpected error occurred invoking '{methodName}' on the server. InvalidOperationException: BOOM!", message.Error);
 
                 // kill the connection
                 client.Dispose();
@@ -465,7 +532,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.SendInvocationAsync(nameof(MethodHub.ValueMethod), nonBlocking: true).OrTimeout();
 
@@ -488,7 +555,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync(nameof(MethodHub.VoidMethod)).OrTimeout()).Result;
 
@@ -510,7 +577,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync("RenamedMethod").OrTimeout()).Result;
 
@@ -533,7 +600,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync("RenamedVirtualMethod").OrTimeout()).Result;
 
@@ -559,7 +626,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient(synchronousCallbacks: true))
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 // This invocation should be completely synchronous
                 await client.SendInvocationAsync(methodName, nonBlocking: true).OrTimeout();
@@ -567,8 +634,9 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                 // kill the connection
                 client.Dispose();
 
-                // Nothing should have been written
+                // only thing written should be close message
-                Assert.False(client.Connection.Application.Input.TryRead(out var buffer));
+                var closeMessage = await client.ReadAsync().OrTimeout();
+                Assert.IsType<CloseMessage>(closeMessage);
 
                 await endPointTask.OrTimeout();
             }
@@ -583,7 +651,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync(nameof(MethodHub.ConcatString), (byte)32, 42, 'm', "string").OrTimeout()).Result;
 
@@ -605,7 +673,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync(nameof(InheritedHub.BaseMethod), "string").OrTimeout()).Result;
 
@@ -627,7 +695,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = (await client.InvokeAsync(nameof(InheritedHub.VirtualMethod), 10).OrTimeout()).Result;
 
@@ -649,7 +717,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = await client.InvokeAsync(nameof(MethodHub.OnDisconnectedAsync)).OrTimeout();
 
@@ -687,7 +755,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = await client.InvokeAsync(nameof(MethodHub.StaticMethod)).OrTimeout();
 
@@ -709,7 +777,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = await client.InvokeAsync(nameof(MethodHub.ToString)).OrTimeout();
                 Assert.Equal("Unknown hub method 'ToString'", result.Error);
@@ -739,7 +807,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 var result = await client.InvokeAsync(nameof(MethodHub.Dispose)).OrTimeout();
 
@@ -761,8 +829,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -796,8 +864,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -831,8 +899,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -871,8 +939,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -891,8 +959,9 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 // kill the connections
                 firstClient.Dispose();
+                secondClient.Dispose();
 
-                await firstEndPointTask.OrTimeout();
+                await Task.WhenAll(firstEndPointTask, secondEndPointTask).OrTimeout();
             }
         }
 
@@ -906,9 +975,9 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var secondClient = new TestClient())
             using (var thirdClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
-                Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
+                Task thirdEndPointTask = await thirdClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).OrTimeout();
 
@@ -949,9 +1018,9 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var secondClient = new TestClient())
             using (var thirdClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
-                Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
+                Task thirdEndPointTask = await thirdClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).OrTimeout();
 
@@ -994,9 +1063,9 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var secondClient = new TestClient(addClaimId: true))
             using (var thirdClient = new TestClient(addClaimId: true))
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
-                Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
+                Task thirdEndPointTask = await thirdClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected, thirdClient.Connected).OrTimeout();
 
@@ -1038,8 +1107,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1079,8 +1148,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1129,8 +1198,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1177,8 +1246,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1217,7 +1286,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointTask = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.SendInvocationAsync(nameof(MethodHub.GroupRemoveMethod), "testGroup").OrTimeout();
 
@@ -1237,8 +1306,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient(addClaimId: true))
             using (var secondClient = new TestClient(addClaimId: true))
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1268,8 +1337,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1298,8 +1367,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var firstClient = new TestClient())
             using (var secondClient = new TestClient())
             {
-                Task firstEndPointTask = endPoint.OnConnectedAsync(firstClient.Connection);
+                Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
-                Task secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
+                Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
 
                 await Task.WhenAll(firstClient.Connected, secondClient.Connected).OrTimeout();
 
@@ -1333,10 +1402,10 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             {
                 client.Connection.SupportedFormats = protocol.TransferFormat;
 
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 // Wait for a connection, or for the endpoint to fail.
-                await client.Connected.OrThrowIfOtherFails(endPointLifetime).OrTimeout();
+                await client.Connected.OrThrowIfOtherFails(endPointTask).OrTimeout();
 
                 var messages = await client.StreamAsync(method, 4).OrTimeout();
 
@@ -1349,7 +1418,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1361,7 +1430,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1378,7 +1447,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1390,7 +1459,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1403,7 +1472,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1416,8 +1485,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var client1 = new TestClient(protocol: new JsonHubProtocol()))
             using (var client2 = new TestClient(protocol: new MessagePackHubProtocol()))
             {
-                var endPointLifetime1 = endPoint.OnConnectedAsync(client1.Connection);
+                Task firstEndPointTask = await client1.ConnectAsync(endPoint);
-                var endPointLifetime2 = endPoint.OnConnectedAsync(client2.Connection);
+                Task secondEndPointTask = await client2.ConnectAsync(endPoint);
 
                 await client1.Connected.OrTimeout();
                 await client2.Connected.OrTimeout();
@@ -1439,8 +1508,8 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                 client1.Dispose();
                 client2.Dispose();
 
-                await endPointLifetime1.OrTimeout();
+                await firstEndPointTask.OrTimeout();
-                await endPointLifetime2.OrTimeout();
+                await secondEndPointTask.OrTimeout();
             }
         }
 
@@ -1481,7 +1550,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1491,7 +1560,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1515,7 +1584,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var client = new TestClient())
             {
                 client.Connection.User.AddIdentity(new ClaimsIdentity(new[] { new Claim(ClaimTypes.NameIdentifier, "name") }));
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1525,7 +1594,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1549,7 +1618,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1564,7 +1633,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1576,7 +1645,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1591,7 +1660,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1613,7 +1682,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             using (var client = new TestClient(synchronousCallbacks: false, protocol: new MessagePackHubProtocol(msgPackOptions)))
             {
                 client.Connection.SupportedFormats = TransferFormat.Binary;
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1628,7 +1697,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1643,7 +1712,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
             {
                 var httpContext = new DefaultHttpContext();
                 client.Connection.SetHttpContext(httpContext);
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1652,7 +1721,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1665,7 +1734,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient())
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection);
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1674,7 +1743,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1686,7 +1755,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient(false, new JsonHubProtocol()))
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection).OrTimeout();
+                Task endPointTask = await client.ConnectAsync(endPoint);
                 await client.Connected.OrTimeout();
 
                 // Send a ping
@@ -1698,7 +1767,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
             }
         }
 
@@ -1712,7 +1781,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient(false, new JsonHubProtocol()))
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection).OrTimeout();
+                Task endPointTask = await client.ConnectAsync(endPoint);
 
                 await client.Connected.OrTimeout();
 
@@ -1727,7 +1796,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                 // Shut down
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
 
                 client.Connection.Transport.Output.Complete();
 
@@ -1753,7 +1822,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests
 
             using (var client = new TestClient(false, new JsonHubProtocol()))
             {
-                var endPointLifetime = endPoint.OnConnectedAsync(client.Connection).OrTimeout();
+                Task endPointTask = await client.ConnectAsync(endPoint);
                 await client.Connected.OrTimeout();
 
                 // Wait 500 ms, but make sure to yield some time up to unblock concurrent threads
@@ -1769,34 +1838,80 @@ namespace Microsoft.AspNetCore.SignalR.Tests
                 // Shut down
                 client.Dispose();
 
-                await endPointLifetime.OrTimeout();
+                await endPointTask.OrTimeout();
 
                 client.Connection.Transport.Output.Complete();
 
-                // We should have all pings
+                // We should have all pings (and close message)
                 HubMessage message;
-                var counter = 0;
+                var pingCounter = 0;
+                var hasCloseMessage = false;
                 while ((message = await client.ReadAsync().OrTimeout()) != null)
                 {
-                    counter += 1;
+                    if (hasCloseMessage)
-                    Assert.IsType<PingMessage>(message);
+                    {
+                        Assert.True(false, "Received message after close");
+                    }
+
+                    switch (message)
+                    {
+                        case PingMessage _:
+                            pingCounter += 1;
+                            break;
+                        case CloseMessage _:
+                            hasCloseMessage = true;
+                            break;
+                        default:
+                            Assert.True(false, "Unexpected message type: " + message.GetType().Name);
+                            break;
+                    }
                 }
-                Assert.InRange(counter, 1, Int32.MaxValue);
+                Assert.InRange(pingCounter, 1, Int32.MaxValue);
             }
         }
 
         [Fact]
-        public async Task NegotiatingFailsIfMsgPackRequestedOverTextOnlyTransport()
+        public async Task EndingConnectionSendsCloseMessageWithNoError()
         {
-            var serviceProvider = HubEndPointTestUtils.CreateServiceProvider(services =>
+            var serviceProvider = HubEndPointTestUtils.CreateServiceProvider();
-                services.Configure<HubOptions>(options =>
-                    options.KeepAliveInterval = TimeSpan.FromMilliseconds(100)));
             var endPoint = serviceProvider.GetService<HubEndPoint<MethodHub>>();
 
-            using (var client = new TestClient(false, new MessagePackHubProtocol()))
+            using (var client = new TestClient(false, new JsonHubProtocol()))
             {
-                client.Connection.SupportedFormats = TransferFormat.Text;
+                Task endPointTask = await client.ConnectAsync(endPoint);
-                var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => endPoint.OnConnectedAsync(client.Connection).OrTimeout());
+
+                await client.Connected.OrTimeout();
+
+                // Shut down
+                client.Dispose();
+
+                await endPointTask.OrTimeout();
+
+                client.Connection.Transport.Output.Complete();
+
+                var message = await client.ReadAsync().OrTimeout();
+
+                var closeMessage = Assert.IsType<CloseMessage>(message);
+                Assert.Null(closeMessage.Error);
+            }
+        }
+
+        [Fact]
+        public async Task ErrorInHubOnConnectSendsCloseMessageWithError()
+        {
+            var serviceProvider = HubEndPointTestUtils.CreateServiceProvider();
+            var endPoint = serviceProvider.GetService<HubEndPoint<OnConnectedThrowsHub>>();
+
+            using (var client = new TestClient(false, new JsonHubProtocol()))
+            {
+                Task endPointTask = await client.ConnectAsync(endPoint);
+
+                var message = await client.ReadAsync().OrTimeout();
+
+                var closeMessage = Assert.IsType<CloseMessage>(message);
+                Assert.Equal("Connection closed with an error. InvalidOperationException: Hub OnConnected failed.", closeMessage.Error);
+
+                await endPointTask.OrTimeout();
             }
         }
 

test/Microsoft.AspNetCore.SignalR.Tests/Internal/DefaultHubProtocolResolverTests.cs

@@ -79,30 +79,13 @@ namespace Microsoft.AspNetCore.SignalR.Common.Protocol.Tests
         }
 
         [Fact]
-        public void DefaultHubProtocolResolverThrowsForNotSupportedProtocol()
+        public void DefaultHubProtocolResolverReturnsNullForNotSupportedProtocol()
         {
             var connection = new Mock<ConnectionContext>();
             connection.Setup(m => m.Features).Returns(new FeatureCollection());
             var mockConnection = new Mock<HubConnectionContext>(connection.Object, TimeSpan.FromSeconds(30), NullLoggerFactory.Instance) { CallBase = true };
             var resolver = new DefaultHubProtocolResolver(AllProtocols, NullLogger<DefaultHubProtocolResolver>.Instance);
-            var exception = Assert.Throws<NotSupportedException>(
+            Assert.Null(resolver.GetProtocol("notARealProtocol", AllProtocolNames, mockConnection.Object));
-                () => resolver.GetProtocol("notARealProtocol", AllProtocolNames, mockConnection.Object));
-
-            Assert.Equal("The protocol 'notARealProtocol' is not supported.", exception.Message);
-        }
-
-        [Theory]
-        [MemberData(nameof(HubProtocols))]
-        public void DefaultHubProtocolResolverThrowsWhenNoProtocolsAreSupported(IHubProtocol protocol)
-        {
-            var connection = new Mock<ConnectionContext>();
-            connection.Setup(m => m.Features).Returns(new FeatureCollection());
-            var mockConnection = new Mock<HubConnectionContext>(connection.Object, TimeSpan.FromSeconds(30), NullLoggerFactory.Instance) { CallBase = true };
-            var supportedProtocols= new List<string>();
-            var resolver = new DefaultHubProtocolResolver(AllProtocols, NullLogger<DefaultHubProtocolResolver>.Instance);
-            var exception = Assert.Throws<NotSupportedException>(
-                () => resolver.GetProtocol(protocol.Name, supportedProtocols, mockConnection.Object));
-            Assert.Equal($"The protocol '{protocol.Name}' is not supported.", exception.Message);
         }
 
         [Fact]