Support for selective message timeouts (#379)

packages/sdk/src/adapters/multipeer/client.ts

@@ -19,6 +19,7 @@ import filterEmpty from '../../utils/filterEmpty';
 export type QueuedMessage = {
 	message: Message;
 	promise?: ExportedPromise;
+	timeoutSeconds?: number;
 };
 
 /**
@@ -125,15 +126,15 @@ export class Client extends EventEmitter {
 		}
 	}
 
-	public queueMessage(message: Message, promise?: ExportedPromise) {
+	public queueMessage(message: Message, promise?: ExportedPromise, timeoutSeconds?: number) {
 		const rule = Rules[message.payload.type] || MissingRule;
 		const beforeQueueMessageForClient = rule.client.beforeQueueMessageForClient || (() => message);
 		message = beforeQueueMessageForClient(this.session, this, message, promise);
 		if (message) {
 			// tslint:disable-next-line:max-line-length
 			log.verbose('network', `Client ${this.id.substr(0, 8)} queue id:${message.id.substr(0, 8)}, type:${message.payload.type}`);
 			log.verbose('network-content', JSON.stringify(message, (key, value) => filterEmpty(value)));
-			this.queuedMessages.push({ message, promise });
+			this.queuedMessages.push({ message, promise, timeoutSeconds });
 		}
 	}
 

packages/sdk/src/adapters/multipeer/protocols/clientExecution.ts

@@ -4,8 +4,10 @@
  */
 
 import { Client, ClientDesyncPreprocessor } from '..';
+import { MissingRule, Rules } from '..';
 import { Message } from '../../..';
 import * as Protocols from '../../../protocols';
+import { ExportedPromise } from '../../../utils/exportedPromise';
 
 /**
  * @hidden
@@ -20,10 +22,6 @@ export class ClientExecution extends Protocols.Protocol implements Protocols.Mid
 
 	constructor(private client: Client) {
 		super(client.conn);
-		// Set timeout a little shorter than the app/session connection, ensuring we don't
-		// cause an app/session message timeout - which is not a supported scenario (there
-		// is no reconnect).
-		this.timeoutSeconds = Protocols.DefaultConnectionTimeoutSeconds * 2 / 3;
 		this.heartbeat = new Protocols.Heartbeat(this);
 		this.beforeRecv = this.beforeRecv.bind(this);
 		// Behave like a server-side endpoint (send heartbeats, measure connection quality)
@@ -34,6 +32,13 @@ export class ClientExecution extends Protocols.Protocol implements Protocols.Mid
 		this.use(this);
 	}
 
+	/** @override */
+	public sendMessage(message: Message, promise?: ExportedPromise, timeoutSeconds?: number) {
+		// Apply timeout to messages going to the client.
+		const rule = Rules[message.payload.type] || MissingRule;
+		super.sendMessage(message, promise, rule.client.timeoutSeconds);
+	}
+
 	public startListening() {
 		super.startListening();
 		if (!this.heartbeatTimer) {

packages/sdk/src/adapters/multipeer/protocols/clientHandshake.ts

@@ -3,9 +3,12 @@
  * Licensed under the MIT License.
  */
 
+import { MissingRule, Rules } from '..';
 import { Client } from '../../..';
+import { Message } from '../../..';
 import { Handshake } from '../../../protocols/handshake';
 import { OperatingModel } from '../../../types/network/operatingModel';
+import { ExportedPromise } from '../../../utils/exportedPromise';
 
 /**
  * @hidden
@@ -17,4 +20,11 @@ export class ClientHandshake extends Handshake {
 	constructor(private client: Client, sessionId: string) {
 		super(client.conn, sessionId, OperatingModel.PeerAuthoritative);
 	}
+
+	/** @override */
+	public sendMessage(message: Message, promise?: ExportedPromise, timeoutSeconds?: number) {
+		// Apply timeout to messages going to the client.
+		const rule = Rules[message.payload.type] || MissingRule;
+		super.sendMessage(message, promise, rule.client.timeoutSeconds);
+	}
 }

packages/sdk/src/adapters/multipeer/protocols/clientStartup.ts

@@ -4,8 +4,11 @@
  */
 
 import { Client } from '..';
+import { MissingRule, Rules } from '..';
+import { Message } from '../../..';
 import * as Protocols from '../../../protocols';
 import * as Payloads from '../../../types/network/payloads';
+import { ExportedPromise } from '../../../utils/exportedPromise';
 
 export class ClientStartup extends Protocols.Protocol {
 	/** @override */
@@ -23,6 +26,13 @@ export class ClientStartup extends Protocols.Protocol {
 		}
 	}
 
+	/** @override */
+	public sendMessage(message: Message, promise?: ExportedPromise, timeoutSeconds?: number) {
+		// Apply timeout to messages going to the client.
+		const rule = Rules[message.payload.type] || MissingRule;
+		super.sendMessage(message, promise, rule.client.timeoutSeconds);
+	}
+
 	/**
 	 * @hidden
 	 */

packages/sdk/src/adapters/multipeer/protocols/clientSync.ts

@@ -59,17 +59,17 @@ export class ClientSync extends Protocols.Protocol {
 	 * @override
 	 * Handle the outgoing message according to the synchronization rules specified for this payload.
 	 */
-	public sendMessage(message: Message, promise?: ExportedPromise) {
+	public sendMessage(message: Message, promise?: ExportedPromise, timeoutSeconds?: number) {
 		message.id = message.id || UUID();
 		const handling = this.handlingForMessage(message);
 		// tslint:disable-next-line:switch-default
 		switch (handling) {
 			case 'allow': {
-				super.sendMessage(message, promise);
+				super.sendMessage(message, promise, timeoutSeconds);
 				break;
 			}
 			case 'queue': {
-				this.client.queueMessage(message, promise);
+				this.client.queueMessage(message, promise, timeoutSeconds);
 				break;
 			}
 			case 'ignore': {
@@ -327,7 +327,7 @@ export class ClientSync extends Protocols.Protocol {
 				break;
 			}
 			for (const queuedMessage of queuedMessages) {
-				this.sendMessage(queuedMessage.message, queuedMessage.promise);
+				this.sendMessage(queuedMessage.message, queuedMessage.promise, queuedMessage.timeoutSeconds);
 			}
 			await this.drainPromises();
 		} while (true);

packages/sdk/src/adapters/multipeer/rules.ts

@@ -66,6 +66,12 @@ export type Rule = {
 	 * Message preprocessing applied by the Client class.
 	 */
 	client: {
+		/**
+		 * If non-zero, a timeout will be set for this message. If we don't receive a reply before the
+		 * timeout expires, the client connection will be closed. Only applicable to messages expecting
+		 * replies.
+		 */
+		timeoutSeconds: number;
 		/**
 		 * Called before a message is queued for later delivery to a client.
 		 * @param session The current session.
@@ -129,6 +135,7 @@ export const DefaultRule: Rule = {
 		after: 'allow'
 	},
 	client: {
+		timeoutSeconds: 0,
 		beforeQueueMessageForClient: (
 			session: Session, client: Client, message: any, promise: ExportedPromise) => {
 			return message;
@@ -585,7 +592,13 @@ export const Rules: { [id in Payloads.PayloadType]: Rule } = {
 	'engine2app-rpc': ClientOnlyRule,
 
 	// ========================================================================
-	'handshake': ClientOnlyRule,
+	'handshake': {
+		...ClientOnlyRule,
+		client: {
+			...ClientOnlyRule.client,
+			timeoutSeconds: 30
+		},
+	},
 
 	// ========================================================================
 	'handshake-complete': ClientOnlyRule,
@@ -610,6 +623,10 @@ export const Rules: { [id in Payloads.PayloadType]: Rule } = {
 			during: 'allow',
 			after: 'allow',
 		},
+		client: {
+			...DefaultRule.client,
+			timeoutSeconds: 30
+		}
 	},
 
 	// ========================================================================
@@ -1004,7 +1021,6 @@ export const Rules: { [id in Payloads.PayloadType]: Rule } = {
 				return message;
 			}
 		}
-
 	},
 
 	// ========================================================================
@@ -1015,6 +1031,10 @@ export const Rules: { [id in Payloads.PayloadType]: Rule } = {
 			before: 'error',
 			during: 'allow',
 			after: 'error'
+		},
+		client: {
+			...DefaultRule.client,
+			timeoutSeconds: 30
 		}
 	},
 

packages/sdk/src/protocols/protocol.ts

@@ -12,22 +12,12 @@ import { ExportedPromise } from '../utils/exportedPromise';
 import filterEmpty from '../utils/filterEmpty';
 import { Middleware } from './middleware';
 
-// tslint:disable:variable-name
-/**
- * The amount of time to wait for a reply message before closing the connection.
- * Set to zero to disable timeouts.
- */
-export let DefaultConnectionTimeoutSeconds = 30;
-// tslint:enable:variable-name
-
 /**
  * @hidden
  * Class to handle sending and receiving messages with a client.
  */
 export class Protocol extends EventEmitter {
 	private middlewares: Middleware[] = [];
-	// tslint:disable-next-line:variable-name
-	private _timeoutSeconds = DefaultConnectionTimeoutSeconds;
 
 	private promise: Promise<void>;
 	private promiseResolve: (value?: void | PromiseLike<void>) => void;
@@ -37,9 +27,6 @@ export class Protocol extends EventEmitter {
 	public get promises() { return this.conn.promises; }
 	public get name() { return this.constructor.name; }
 
-	public get timeoutSeconds() { return this._timeoutSeconds; }
-	public set timeoutSeconds(value) { this._timeoutSeconds = value; }
-
 	// tslint:disable-next-line:variable-name
 	constructor(private _conn: Connection) {
 		super();
@@ -84,7 +71,7 @@ export class Protocol extends EventEmitter {
 		this.sendMessage({ payload }, promise);
 	}
 
-	public sendMessage(message: Message, promise?: ExportedPromise) {
+	public sendMessage(message: Message, promise?: ExportedPromise, timeoutSeconds?: number) {
 		message.id = message.id || UUID();
 
 		// Run message through all the middlewares
@@ -102,14 +89,14 @@ export class Protocol extends EventEmitter {
 		}
 
 		const setReplyTimeout = () => {
-			if (this.timeoutSeconds > 0) {
+			if (timeoutSeconds > 0) {
 				return setTimeout(() => {
 					// tslint:disable-next-line:max-line-length
 					const reason = `${this.name} timed out awaiting response for ${message.payload.type}, id:${message.id}.`;
 					log.error('network', reason);
 					this.rejectPromiseForMessage(message.id, reason);
 					this.conn.close();
-				}, this.timeoutSeconds * 1000);
+				}, timeoutSeconds * 1000);
 			}
 		};