huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Rename negotiate to handshake, add handshake response, add close frame (#1591)

This commit is contained in:
James Newton-King 2018-03-20 17:40:22 +13:00 committed by GitHub
parent 71fd4ffd8a
commit 4f10560087
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
46 changed files with 1406 additions and 793 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

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

8
client-ts/FunctionalTests/ts/HubConnectionTests.ts
View File

@ -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();

14
client-ts/signalr-protocol-msgpack/src/MessagePackHubProtocol.ts
View File

@ -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.");

215
client-ts/signalr/spec/HubConnection.spec.ts
View File

@ -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);
@ -562,17 +727,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]>;

5
client-ts/signalr/spec/tsconfig.json
View File

@ -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": [
"./**/*",

120
client-ts/signalr/src/HubConnection.ts
View File

@ -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);
@ -63,36 +64,104 @@ export class HubConnection {
clearTimeout(this.timeoutHandle);
}
// Parse the messages
const messages = this.protocol.parseMessages(data);
if (!this.receivedHandshakeResponse) {
data = this.processHandshakeResponse(data);
this.receivedHandshakeResponse = true;
}
for (const message of messages) {
switch (message.type) {
case MessageType.Invocation:
this.invokeClientMethod(message);
break;
case MessageType.StreamItem:
case MessageType.Completion:
const callback = this.callbacks.get(message.invocationId);
if (callback != null) {
if (message.type === MessageType.Completion) {
this.callbacks.delete(message.invocationId);
// Data may have all been read when processing handshake response
if (data) {
// Parse the messages
const messages = this.protocol.parseMessages(data);
for (const message of messages) {
switch (message.type) {
case MessageType.Invocation:
this.invokeClientMethod(message);
break;
case MessageType.StreamItem:
case MessageType.Completion:
const callback = this.callbacks.get(message.invocationId);
if (callback != null) {
if (message.type === MessageType.Completion) {
this.callbacks.delete(message.invocationId);
}
callback(message);
}
callback(message);
}
break;
case MessageType.Ping:
// Don't care about pings
break;
default:
this.logger.log(LogLevel.Warning, "Invalid message type: " + data);
break;
break;
case MessageType.Ping:
// Don't care about pings
break;
case MessageType.Close:
this.logger.log(LogLevel.Information, "Close message received from server.");
this.connection.stop(message.error ? new Error("Server returned an error on close: " + message.error) : null);
break;
default:
this.logger.log(LogLevel.Warning, "Invalid message type: " + 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
@ -133,11 +202,14 @@ export class HubConnection {
}
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}'.`);

16
client-ts/signalr/src/IHubProtocol.ts
View File

@ -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;
}

3
client-ts/signalr/src/TextMessageFormat.ts
View File

@ -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}`;

98
specs/HubProtocol.md
View File

@ -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)
* 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)
* 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 `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.
* `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.
* `StreamItem` Message - Indicates individual items of streamed response data from a previous Invocation message.
* `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
* `CancelInvocation` Message - Sent by the client to cancel a streaming invocation on the server.
* `Ping` Message - Sent by either party to check if the connection is active.
| Message Name | Sender | Description |
| ------------------ | -------------- | ------------------------------------------------------------------------------------------------------------------------------ |
| `HandshakeRequest` | Client | Sent by the client to agree on the message format. |
| `HandshakeResponse` | Server | Sent by the server as an acknowledgment of the previous `HandshakeRequest` message. Contains an error if the handshake failed. |
| `Close` | Callee, Caller | Sent by the server when a connection is closed. Contains an error if the connection was closed because of an error. |
| `Invocation` | Caller | Indicates a request to invoke a particular method (the Target) with provided Arguments on the remote endpoint. |
| `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,

40
src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.Log.cs
View File

@ -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 =
LoggerMessage.Define(LogLevel.Debug, new EventId(28, "SendingHubNegotiate"), "Sending Hub Negotiation.");
private static readonly Action<ILogger, Exception> _sendingHubHandshake =
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);
}
}
}
}

87
src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.cs
View File

@ -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);
NegotiationProtocol.WriteMessage(new NegotiationMessage(_protocol.Name), memoryStream);
Log.SendingHubHandshake(_logger);
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)
{

20
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/CloseMessage.cs Normal file
View File

@ -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;
}
}
}

128
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HandshakeProtocol.cs Normal file
View File

@ -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;
}
}
}

4
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/NegotiationMessage.cs → src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HandshakeRequestMessage.cs
View File

@ -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;
}

17
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HandshakeResponseMessage.cs Normal file
View File

@ -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;
}
}
}

1
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/HubProtocolConstants.cs
View File

@ -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;
}
}

30
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/JsonHubProtocol.cs
View File

@ -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];

10
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/JsonUtils.cs
View File

@ -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];

80
src/Microsoft.AspNetCore.SignalR.Common/Internal/Protocol/NegotiationProtocol.cs
View File

@ -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);
}
}
}

6
src/Microsoft.AspNetCore.SignalR.Common/Properties/AssemblyInfo.cs Normal file
View File

@ -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")]

82
src/Microsoft.AspNetCore.SignalR.Core/HubConnectionContext.cs
View File

@ -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 =
LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, "UsingHubProtocol"), "Using HubProtocol '{Protocol}'.");
private static readonly Action<ILogger, string, Exception> _handshakeComplete =
LoggerMessage.Define<string>(LogLevel.Information, new EventId(1, "HandshakeComplete"), "Completed connection handshake. Using HubProtocol '{Protocol}'.");
private static readonly Action<ILogger, Exception> _negotiateCanceled =
LoggerMessage.Define(LogLevel.Debug, new EventId(2, "NegotiateCanceled"), "Negotiate was canceled.");
private static readonly Action<ILogger, Exception> _handshakeCanceled =
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);
}
}
}

44
src/Microsoft.AspNetCore.SignalR.Core/HubEndPoint.cs
View File

@ -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);
}
}
}
}

4
src/Microsoft.AspNetCore.SignalR.Core/HubOptions.cs
View File

@ -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;

6
src/Microsoft.AspNetCore.SignalR.Core/HubOptionsSetup.cs
View File

@ -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)

2
src/Microsoft.AspNetCore.SignalR.Core/HubOptionsSetup`T.cs
View File

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

4
src/Microsoft.AspNetCore.SignalR.Core/Internal/DefaultHubProtocolResolver.cs
View File

@ -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

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

@ -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);

3
src/Microsoft.AspNetCore.Sockets.Abstractions/ConnectionMetadata.cs
View File

@ -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>())
{
}

20
src/Microsoft.AspNetCore.Sockets.Http/HttpConnectionDispatcher.cs
View File

@ -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)

25
test/Common/ChannelExtensions.cs
View File

@ -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;
}
}
}

112
test/Common/ServerFixture.cs
View File

@ -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;
}
}
}
}

213
test/Common/TestClient.cs
View File

@ -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);
}
}
}
}

24
test/Common/TestHelpers.cs
View File

@ -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;
}
}
}

4
test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs
View File

@ -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)
{

15
test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/Hubs.cs
View File

@ -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]);
}

4
test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionExtensionsTests.cs
View File

@ -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

96
test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionProtocolTests.cs
View File

@ -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
await connection.ReadSentTextMessageAsync().OrTimeout();
// Ignore handshake message
await connection.ReadHandshakeAndSendResponseAsync().OrTimeout();
// Send an invocation
var invokeTask = hubConnection.InvokeAsync("Foo");

11
test/Microsoft.AspNetCore.SignalR.Client.Tests/TestConnection.cs
View File

@ -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();

81
test/Microsoft.AspNetCore.SignalR.Common.Tests/Internal/Protocol/HandshakeProtocolTests.cs Normal file
View File

@ -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);
}
}
}

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

@ -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);
}
}
}

6
test/Microsoft.AspNetCore.SignalR.Tests.Utils/Microsoft.AspNetCore.SignalR.Tests.Utils.csproj
View File

@ -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" />

64
test/Microsoft.AspNetCore.SignalR.Tests.Utils/TestClient.cs
View File

@ -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 Queue<HubMessage> _messages = new Queue<HubMessage>();
private readonly CancellationTokenSource _cts;
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);
Connection.Application.Output.WriteAsync(memoryStream.ToArray());
using (var memoryStream = new MemoryStream())
{
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 (_protocol.TryParseMessages(result.Buffer.ToArray(), _invocationBinder, messages))
if (!isHandshake)
{
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

30
test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs
View File

@ -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
var receivedData = await receiveTcs.Task;
// Big timeout here because it can take a while to receive all the bytes
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>(
async () => await ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType.WebSockets));
Assert.Equal("Websocket closed with error: InternalServerError.", exception.Message);
try
{
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>(
async () => await ServerClosesConnectionWithErrorIfHubCannotBeCreated(TransportType.LongPolling));
Assert.Equal("Response status code does not indicate success: 500 (Internal Server Error).", exception.Message);
try
{
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.
}

2
test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTestUtils/Hubs.cs
View File

@ -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)
{

351
test/Microsoft.AspNetCore.SignalR.Tests/HubEndpointTests.cs
View File

@ -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
client.Connection.Transport.Input.TryRead(out var item);
client.Connection.Transport.Input.AdvanceTo(item.Buffer.End);
var endPointTask = endPoint.OnConnectedAsync(client.Connection);
Task endPointTask = await client.ConnectAsync(endPoint, false, false);
// 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
client.Connection.Transport.Input.TryRead(out var item);
client.Connection.Transport.Input.AdvanceTo(item.Buffer.End);
Task endPointTask = await client.ConnectAsync(endPoint, false);
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);
Assert.Equal("Hub OnConnected failed.", exception.Message);
await endPointTask.OrTimeout();
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
Assert.False(client.Connection.Application.Input.TryRead(out var buffer));
// only thing written should be close message
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task thirdEndPointTask = endPoint.OnConnectedAsync(thirdClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
Task secondEndPointTask = await secondClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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 secondEndPointTask = endPoint.OnConnectedAsync(secondClient.Connection);
Task firstEndPointTask = await firstClient.ConnectAsync(endPoint);
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);
var endPointLifetime2 = endPoint.OnConnectedAsync(client2.Connection);
Task firstEndPointTask = await client1.ConnectAsync(endPoint);
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 endPointLifetime2.OrTimeout();
await firstEndPointTask.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;
Assert.IsType<PingMessage>(message);
if (hasCloseMessage)
{
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 =>
services.Configure<HubOptions>(options =>
options.KeepAliveInterval = TimeSpan.FromMilliseconds(100)));
var serviceProvider = HubEndPointTestUtils.CreateServiceProvider();
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;
var ex = await Assert.ThrowsAsync<InvalidOperationException>(() => endPoint.OnConnectedAsync(client.Connection).OrTimeout());
Task endPointTask = await client.ConnectAsync(endPoint);
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();
}
}

21
test/Microsoft.AspNetCore.SignalR.Tests/Internal/DefaultHubProtocolResolverTests.cs
View File

@ -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>(
() => 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);
Assert.Null(resolver.GetProtocol("notARealProtocol", AllProtocolNames, mockConnection.Object));
}
[Fact]