diff --git a/packages/lib/src/data/messages_core.ts b/packages/lib/src/data/messages_core.ts index 305905e4..c4145a47 100644 --- a/packages/lib/src/data/messages_core.ts +++ b/packages/lib/src/data/messages_core.ts @@ -114,6 +114,24 @@ export class Address { return `${this.type}:${this.id}` as const; } + /** + * Returns this address with any request id removed + * + * A request id identifies one request in flight and is set on the + * source address of a request so that its response can be matched back + * up with it. It is meaningless on anything else, so an address taken + * from a request has to be stripped of it before it is used as the + * destination of an event. + * + * @returns this address without a request id. + */ + withoutRequestId() { + if (this.requestId === undefined) { + return this; + } + return new Address(this.type, this.id); + } + /** * Returns true if this address targets the given destination * @param dst - Destination to check diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index 72b2b24b..7e047f34 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -345,6 +345,9 @@ export class SubscriptionController { } const addressIndex = src.addressIndex(); const subscriber = eventData.subscriptions.get(addressIndex); + // src is the source of a request and carries its request id, which + // has no meaning on the events sent back to the subscriber. + const subscriberDst = src.withoutRequestId(); switch (request.action) { case "unsubscribe": if (!subscriber) { @@ -358,7 +361,9 @@ export class SubscriptionController { case "subscribe": if (!subscriber) { - eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + eventData.subscriptions.set( + addressIndex, { link: link, dst: subscriberDst, filters: request.filters } + ); } else { subscriber.filters.union(request.filters); } @@ -371,7 +376,9 @@ export class SubscriptionController { } eventData.subscriptions.delete(addressIndex); } else if (!subscriber) { - eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + eventData.subscriptions.set( + addressIndex, { link: link, dst: subscriberDst, filters: request.filters } + ); } else { subscriber.filters = request.filters; } @@ -384,7 +391,7 @@ export class SubscriptionController { if (eventData.subscriptionUpdate) { const eventReplay = await eventData.subscriptionUpdate(request, src, dst); if (eventReplay) { - link.sendTo(src, eventReplay); + link.sendTo(subscriberDst, eventReplay); return true; } } diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 313b2968..eb4659ed 100644 --- a/test/lib/subscriptions.test.js +++ b/test/lib/subscriptions.test.js @@ -520,6 +520,72 @@ describe("lib/subscriptions", function() { }); }); + describe("subscriber addressing", function() { + // A subscribe arrives as a request, so the source address it is + // handled with carries the request id used to route the response + // back to the caller. That id identifies one request in flight and + // means nothing on an event, see #847. These use a controller of + // their own because a subscription is only ever addressed when it + // is first created, and the shared one is already subscribed. + const connectorData = connectorSetupDate[0]; + + function requestSource(requestId) { + return new lib.Address(connectorData.dst.type, connectorData.dst.id, requestId); + } + + function lastSentMessage() { + return getLink(connectorData.id).connector.sentMessages.at(-1); + } + + it("should strip the request id from broadcast events", async function() { + const ownSubscriptions = new lib.SubscriptionController(mockController); + ownSubscriptions.handle(RegisteredEvent); + await ownSubscriptions.handleRequest( + getLink(connectorData.id), + new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"), + requestSource(213), connectorData.src + ); + + ownSubscriptions.broadcast(new RegisteredEvent()); + await onceConnectorSend(connectorData.id); + + assert.equal(lastSentMessage().name, RegisteredEvent.name); + assert.equal(lastSentMessage().dst.requestId, undefined, "event addressed to a request id"); + assert.deepEqual(lastSentMessage().dst.toJSON(), connectorData.dst.toJSON()); + }); + + it("should strip the request id from the replayed event", async function() { + const ownSubscriptions = new lib.SubscriptionController(mockController); + ownSubscriptions.handle(RegisteredEvent, async () => new RegisteredEvent()); + await ownSubscriptions.handleRequest( + getLink(connectorData.id), + new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe"), + requestSource(417), connectorData.src + ); + + assert.equal(lastSentMessage().name, RegisteredEvent.name); + assert.equal(lastSentMessage().dst.requestId, undefined, "replay addressed to a request id"); + assert.deepEqual(lastSentMessage().dst.toJSON(), connectorData.dst.toJSON()); + }); + + it("should strip the request id when replacing filters creates the subscription", async function() { + const ownSubscriptions = new lib.SubscriptionController(mockController); + ownSubscriptions.handle(RegisteredEvent); + await ownSubscriptions.handleRequest( + getLink(connectorData.id), + new lib.SubscriptionRequest(RegisteredEvent.name, "replace", 0, "foo"), + requestSource(918), connectorData.src + ); + + ownSubscriptions.broadcast(new RegisteredEvent(), "foo"); + await onceConnectorSend(connectorData.id); + + assert.equal(lastSentMessage().name, RegisteredEvent.name); + assert.equal(lastSentMessage().dst.requestId, undefined, "event addressed to a request id"); + assert.deepEqual(lastSentMessage().dst.toJSON(), connectorData.dst.toJSON()); + }); + }); + describe(".unsubscribeLink() / .unsubscribeAddress()", function() { beforeEach(function() { subscriptions.handle(RegisteredEvent);