import { ConnectionClosed } from "./Common" import { IConnection } from "./IConnection" import { TransportType } from "./Transports" import { Subject, Observable } from "./Observable" export { TransportType } from "./Transports" export { HttpConnection } from "./HttpConnection" import { IHubProtocol, MessageType, HubMessage, CompletionMessage, ResultMessage, InvocationMessage, NegotiationMessage } from "./IHubProtocol"; import { JsonHubProtocol } from "./JsonHubProtocol"; import { TextMessageFormat } from "./Formatters" export class HubConnection { private connection: IConnection; private callbacks: Map void>; private methods: Map void>; private id: number; private connectionClosedCallback: ConnectionClosed; private protocol: IHubProtocol; constructor(connection: IConnection) { this.connection = connection; this.connection.onDataReceived = data => { this.onDataReceived(data); }; this.connection.onClosed = (error: Error) => { this.onConnectionClosed(error); } this.callbacks = new Map void>(); this.methods = new Map void>(); this.id = 0; this.protocol = new JsonHubProtocol(); } private onDataReceived(data: any) { // Parse the messages let messages = this.protocol.parseMessages(data); for (var i = 0; i < messages.length; ++i) { var message = messages[i]; switch (message.type) { case MessageType.Invocation: this.invokeClientMethod(message); break; case MessageType.Result: case MessageType.Completion: let callback = this.callbacks.get(message.invocationId); if (callback != null) { callback(message); if (message.type == MessageType.Completion) { this.callbacks.delete(message.invocationId); } } break; default: console.log("Invalid message type: " + data); break; } } } private invokeClientMethod(invocationMessage: InvocationMessage) { let method = this.methods.get(invocationMessage.target); if (method) { method.apply(this, invocationMessage.arguments); if (!invocationMessage.nonblocking) { // TODO: send result back to the server? } } else { console.log(`No client method with the name '${invocationMessage.target}' found.`); } } private onConnectionClosed(error: Error) { let errorCompletionMessage = { type: MessageType.Completion, invocationId: "-1", error: error ? error.message : "Invocation cancelled due to connection being closed.", }; this.callbacks.forEach(callback => { callback(errorCompletionMessage); }); this.callbacks.clear(); if (this.connectionClosedCallback) { this.connectionClosedCallback(error); } } async start(): Promise { await this.connection.start(); await this.connection.send( TextMessageFormat.write( JSON.stringify({ protocol: this.protocol.name()}))); } stop(): void { return this.connection.stop(); } stream(methodName: string, ...args: any[]): Observable { let invocationDescriptor = this.createInvocation(methodName, args); let subject = new Subject(); this.callbacks.set(invocationDescriptor.invocationId, (invocationEvent: CompletionMessage | ResultMessage) => { if (invocationEvent.type === MessageType.Completion) { let completionMessage = invocationEvent; if (completionMessage.error) { subject.error(new Error(completionMessage.error)); } else if (completionMessage.result) { subject.error(new Error("Server provided a result in a completion response to a streamed invocation.")); } else { // TODO: Log a warning if there's a payload? subject.complete(); } } else { subject.next((invocationEvent).item); } }); let message = this.protocol.writeMessage(invocationDescriptor); this.connection.send(message) .catch(e => { subject.error(e); this.callbacks.delete(invocationDescriptor.invocationId); }); return subject; } invoke(methodName: string, ...args: any[]): Promise { let invocationDescriptor = this.createInvocation(methodName, args); let p = new Promise((resolve, reject) => { this.callbacks.set(invocationDescriptor.invocationId, (invocationEvent: CompletionMessage | ResultMessage) => { if (invocationEvent.type === MessageType.Completion) { let completionMessage = invocationEvent; if (completionMessage.error) { reject(new Error(completionMessage.error)); } else { resolve(completionMessage.result); } } else { reject(new Error("Streaming methods must be invoked using HubConnection.stream")) } }); let message = this.protocol.writeMessage(invocationDescriptor); this.connection.send(message) .catch(e => { reject(e); this.callbacks.delete(invocationDescriptor.invocationId); }); }); return p; } on(methodName: string, method: (...args: any[]) => void) { this.methods.set(methodName, method); } set onClosed(callback: ConnectionClosed) { this.connectionClosedCallback = callback; } private createInvocation(methodName: string, args: any[]): InvocationMessage { let id = this.id; this.id++; return { type: MessageType.Invocation, invocationId: id.toString(), target: methodName, arguments: args, nonblocking: false }; } }