diff --git a/CHANGELOG.md b/CHANGELOG.md index 6ece0937..7e03d59a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Version 2.0.0-alpha.16 + +## Changes + +- Fixed message sequences not being renumbered when forwarded between links causing reconnects to send duplicated messages. + + ## Version 2.0.0-alpha.15 ### Features diff --git a/packages/controller/src/ControllerRouter.ts b/packages/controller/src/ControllerRouter.ts index 4679312d..4072c941 100644 --- a/packages/controller/src/ControllerRouter.ts +++ b/packages/controller/src/ControllerRouter.ts @@ -84,7 +84,7 @@ export default class ControllerRouter { if (message.type === "request") { nextHop.forwardRequest(message as lib.MessageRequest, origin); } else { - nextHop.connector.send(message); + nextHop.connector.forward(message); } } else { this.warnUnrouted(message, msg); @@ -107,13 +107,13 @@ export default class ControllerRouter { if (dst.id === lib.Address.host || dst.id === lib.Address.instance) { for (let hostConnection of this.controller.wsServer.hostConnections.values()) { if (hostConnection !== origin && (!plugin || hostConnection.plugins.has(plugin))) { - hostConnection.connector.send(message); + hostConnection.connector.forward(message); } } } else if (dst.id === lib.Address.control) { for (let controlConnection of this.controller.wsServer.controlConnections.values()) { if (controlConnection !== origin) { - controlConnection.connector.send(message); + controlConnection.connector.forward(message); } } } else { diff --git a/packages/host/src/Host.ts b/packages/host/src/Host.ts index b1b9dd41..c2fc5d67 100644 --- a/packages/host/src/Host.ts +++ b/packages/host/src/Host.ts @@ -173,16 +173,16 @@ export class HostRouter { if (dst.id === lib.Address.instance) { for (let instanceConnection of this.host.instanceConnections.values()) { if (instanceConnection !== origin && (!plugin || instanceConnection.plugins.has(plugin))) { - instanceConnection.connector.send(message); + instanceConnection.connector.forward(message); } } if (this.host !== origin && (!plugin || this.host.serverPlugins.has(plugin))) { - this.host.connector.send(message); + this.host.connector.forward(message); } } else if (dst.id === lib.Address.control) { if (this.host !== origin) { if (!plugin || this.host.serverPlugins.has(plugin)) { - this.host.connector.send(message); + this.host.connector.forward(message); } } else { logger.warn(`Received control broadcast of ${(message as lib.MessageEvent).name} from controller.`); @@ -208,7 +208,7 @@ export class HostRouter { } this.host._connectInstance(dst.id).then(instanceConnection => { - instanceConnection.connector.send(message); + instanceConnection.connector.forward(message); }).catch(err => { logger.error(`Error starting instance:\n${err.stack}`, this.host.instanceLogMeta(dst.id)); origin.connector.sendResponseError( @@ -222,7 +222,7 @@ export class HostRouter { if (message.type === "request") { nextHop.forwardRequest(message as lib.MessageRequest, origin); } else { - nextHop.connector.send(message); + nextHop.connector.forward(message); } } catch (err: any) { if (message.type === "request") { diff --git a/packages/lib/src/link/connectors.ts b/packages/lib/src/link/connectors.ts index 2bc15f19..999800ba 100644 --- a/packages/lib/src/link/connectors.ts +++ b/packages/lib/src/link/connectors.ts @@ -33,7 +33,30 @@ export abstract class BaseConnector extends events.EventEmitter { this._seq = 1; } - abstract send(message: libData.Message): void; + protected abstract send(message: libData.MessageRoutable): void; + + forward(message: libData.MessageRoutable) { + const seq = this._seq; + this._seq += 1; + // The message type currently can't be inferred like a tagged union + // TODO fix typing of message so it can be deduced from .type + if (message.type === "request") { + const request = message as libData.MessageRequest; + this.send(new libData.MessageRequest(seq, request.src, request.dst, request.name, request.data)); + } else if (message.type === "response") { + const response = message as libData.MessageResponse; + this.send(new libData.MessageResponse(seq, response.src, response.dst, response.data)); + } else if (message.type === "responseError") { + const error = message as libData.MessageResponseError; + this.send(new libData.MessageResponseError(seq, error.src, error.dst, error.data)); + } else if (message.type === "event") { + const event = message as libData.MessageEvent; + this.send(new libData.MessageEvent(seq, event.src, event.dst, event.name, event.data)); + } else { + throw new Error(`Cannot forward message type ${(message as any).type}`); + } + return seq; + } sendRequest( request: Request, @@ -242,7 +265,7 @@ export abstract class WebSocketBaseConnector extends BaseConnector { * * @param message - Message to send. */ - send(message: libData.MessageRoutable) { + protected send(message: libData.MessageRoutable) { if (!["connected", "resuming"].includes(this._state)) { throw new libErrors.SessionLost("No session"); } @@ -295,6 +318,13 @@ export abstract class WebSocketBaseConnector extends BaseConnector { await this.close(1000, "Disconnect"); } + /** + * Notify other end that the link is ready to disconnect + */ + sendDisconnectReady() { + this._sendInternal(new libData.MessageDisconnect("ready")); + } + /** * True if the connection is established and active */ diff --git a/packages/lib/src/link/link.ts b/packages/lib/src/link/link.ts index 520e5fe4..e3a746c0 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -132,7 +132,8 @@ export class Link { } ).finally( () => { - this.connector.send(new libData.MessageDisconnect("ready")); + // Only WebSocket connectors issue disconnection events. + (this.connector as WebSocketBaseConnector).sendDisconnectReady(); }, ); }); @@ -561,7 +562,7 @@ export class Link { dst: message.dst, }; this._forwardedRequests.set(message.src.index(), pending); - this.connector.send(message); + this.connector.forward(message); } sendEvent(event: Event, dst: libData.Address) { diff --git a/test/mock.js b/test/mock.js index a00729c9..171c5923 100644 --- a/test/mock.js +++ b/test/mock.js @@ -80,6 +80,10 @@ class MockConnector extends lib.BaseConnector { this.sentMessages.push(message); setImmediate(() => this.emit("send", message)); } + + sendDisconnectReady() { + this.send(new lib.MessageDisconnect("ready")); + } } class MockServer extends events.EventEmitter {