diff --git a/packages/controller/src/ControllerRouter.js b/packages/controller/src/ControllerRouter.js index 2b4b1f6e..8bc29fcb 100644 --- a/packages/controller/src/ControllerRouter.js +++ b/packages/controller/src/ControllerRouter.js @@ -84,7 +84,11 @@ class ControllerRouter { } if (nextHop) { - nextHop.connector.send(message); + if (message.type === "request") { + nextHop.forwardRequest(message, origin); + } else { + nextHop.connector.send(message); + } } else { this.warnUnrouted(message, msg); } diff --git a/packages/host/src/Host.js b/packages/host/src/Host.js index 3b7eea31..b616820e 100644 --- a/packages/host/src/Host.js +++ b/packages/host/src/Host.js @@ -151,7 +151,7 @@ class HostRouter { } if (nextHop) { - this.sendMessage(nextHop, message); + this.sendMessage(nextHop, message, origin); } else { this.warnUnrouted(message, msg); } @@ -208,9 +208,13 @@ class HostRouter { }); } - sendMessage(nextHop, message) { + sendMessage(nextHop, message, origin) { try { - nextHop.connector.send(message); + if (message.type === "request") { + nextHop.forwardRequest(message, origin); + } else { + nextHop.connector.send(message); + } } catch (err) { if (message.type === "request") { origin.connector.sendResponseError( diff --git a/packages/lib/data/messages_core.js b/packages/lib/data/messages_core.js index 037e4c88..2b74c125 100644 --- a/packages/lib/data/messages_core.js +++ b/packages/lib/data/messages_core.js @@ -98,6 +98,14 @@ class Address { return `[Address ${type}:${this.id}:${this.requestId}]`; } + /** + * Returns a string suitable for indexing the address in a Map + * @returns {string} unique string for this address + */ + index() { + return `${this.type}:${this.id}:${this.requestId}`; + } + /** * Returns true if this address targets the given destination * @param {module:lib/Data.Address} dst - Destination to check diff --git a/packages/lib/link/link.js b/packages/lib/link/link.js index dd4bf6fe..c21592c7 100644 --- a/packages/lib/link/link.js +++ b/packages/lib/link/link.js @@ -30,6 +30,7 @@ class Link { this._eventHandlers = new Map(); this._eventSnoopers = new Map(); this._pendingRequests = new Map(); + this._forwardedRequests = new Map(); this._nextRequestId = 1; this.handle(libData.PingRequest, () => {}); @@ -65,22 +66,29 @@ class Link { }); connector.on("invalidate", () => { - for (let pending of this._pendingRequests.values()) { - pending.reject(new libErrors.SessionLost("Session Lost")); - } - - this._pendingRequests.clear(); + this._clearPendingRequests(new libErrors.SessionLost("Session Lost")); }); connector.on("close", () => { - for (let pending of this._pendingRequests.values()) { - pending.reject(new libErrors.SessionLost("Session Closed")); - } - - this._pendingRequests.clear(); + this._clearPendingRequests(new libErrors.SessionLost("Session Closed")); }); } + _clearPendingRequests(err) { + for (let pending of this._pendingRequests.values()) { + pending.reject(err); + } + for (let pending of this._forwardedRequests.values()) { + if (pending.origin.connector.hasSession) { + pending.origin.connector.sendResponseError( + new libData.ResponseError(err.message, err.code), pending.src, pending.dst + ); + } + } + this._pendingRequests.clear(); + this._forwardedRequests.clear(); + } + /** * Process a received message on the link * @@ -115,6 +123,9 @@ class Link { } if (!message.dst.addressedTo(this.connector.src)) { + if (message.type === "response" || message.type === "responseError") { + this._forwardedRequests.delete(message.dst.index()); + } this._routeMessage(message, entry); return; } @@ -411,6 +422,24 @@ class Link { return pending.promise; } + forwardRequest(message, origin) { + if (this.validateSent) { + let entry = this.constructor._requestsByName.get(message.name); + if (!entry) { + throw new Error(`Attempt to forward unregistered Request ${message.name}`); + } + entry.requestFromJSON(message.data); + } + + let pending = { + origin, + src: message.src, + dst: message.dst, + }; + this._forwardedRequests.set(message.src.index(), pending); + this.connector.send(message); + } + sendEvent(event, dst) { let entry = this.constructor._eventsByClass.get(event.constructor); if (!entry) {