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..1b3311b6 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -238,7 +238,7 @@ export class SubscriptionRequest { type Subscriber = { link: Link, - dst: Address, + address: Address, filters: SubscriptionFilters, } @@ -296,7 +296,7 @@ export class SubscriptionController { continue; } if (subscriber.filters.intersects(broadcastFilters)) { - subscriber.link.sendTo(subscriber.dst, event); + subscriber.link.sendTo(subscriber.address, event); } } } @@ -343,8 +343,9 @@ export class SubscriptionController { if (!eventData) { throw new Error(`Event ${request.eventName} is not a registered as subscribable`); } - const addressIndex = src.addressIndex(); - const subscriber = eventData.subscriptions.get(addressIndex); + const index = src.addressIndex(); + const address = src.withoutRequestId(); + const subscriber = eventData.subscriptions.get(index); switch (request.action) { case "unsubscribe": if (!subscriber) { @@ -352,13 +353,13 @@ export class SubscriptionController { } subscriber.filters.subtract(request.filters); if (subscriber.filters.isEmpty()) { - eventData.subscriptions.delete(addressIndex); + eventData.subscriptions.delete(index); } break; case "subscribe": if (!subscriber) { - eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + eventData.subscriptions.set(index, { link, address, filters: request.filters }); } else { subscriber.filters.union(request.filters); } @@ -369,9 +370,9 @@ export class SubscriptionController { if (!subscriber) { return false; } - eventData.subscriptions.delete(addressIndex); + eventData.subscriptions.delete(index); } else if (!subscriber) { - eventData.subscriptions.set(addressIndex, { link: link, dst: src, filters: request.filters }); + eventData.subscriptions.set(index, { link, address, filters: request.filters }); } else { subscriber.filters = request.filters; } @@ -384,7 +385,7 @@ export class SubscriptionController { if (eventData.subscriptionUpdate) { const eventReplay = await eventData.subscriptionUpdate(request, src, dst); if (eventReplay) { - link.sendTo(src, eventReplay); + link.sendTo(address, eventReplay); return true; } } diff --git a/test/lib/data/messages_core.js b/test/lib/data/messages_core.js index 3ff28509..2e06c673 100644 --- a/test/lib/data/messages_core.js +++ b/test/lib/data/messages_core.js @@ -16,6 +16,33 @@ describe("lib/data/messages_core", function() { const instanceBr = new Address(Address.broadcast, Address.instance); const controlBr = new Address(Address.broadcast, Address.control); + describe(".withoutRequestId()", () => { + it("returns the same instance when requestId is undefined", () => { + const address = new Address(1, 2); + + assert.strictEqual(address.withoutRequestId(), address); + }); + + it("returns a new instance when requestId is present", () => { + const address = new Address(1, 2, 123); + + const result = address.withoutRequestId(); + + assert.notStrictEqual(result, address); + assert.deepEqual(result, new Address(1, 2)); + }); + + it("does not modify the original address", () => { + const address = new Address(1, 2, 123); + + address.withoutRequestId(); + + assert.equal(address.type, 1); + assert.equal(address.id, 2); + assert.equal(address.requestId, 123); + }); + }); + it("Should correctly resolve addressedTo", function() { assert(controller.addressedTo(controller)); assert(!controller.addressedTo(host1)); diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 313b2968..bd556487 100644 --- a/test/lib/subscriptions.test.js +++ b/test/lib/subscriptions.test.js @@ -614,6 +614,71 @@ describe("lib/subscriptions", function() { ), /unreachable case: invalid action/); }); + describe("addressing", function() { + // A subscribe arrives as a request, so the source address it is + // handled with carries the request id which should not be persisted + 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); + + const message = lastSentMessage(); + assert.equal(message.name, RegisteredEvent.name); + assert.equal(message.dst.requestId, undefined, "event addressed to a request id"); + assert.deepEqual(message.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 + ); + + const message = lastSentMessage(); + assert.equal(message.name, RegisteredEvent.name); + assert.equal(message.dst.requestId, undefined, "replay addressed to a request id"); + assert.deepEqual(message.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); + + const message = lastSentMessage(); + assert.equal(message.name, RegisteredEvent.name); + assert.equal(message.dst.requestId, undefined, "event addressed to a request id"); + assert.deepEqual(message.dst.toJSON(), connectorData.dst.toJSON()); + }); + }); + describe("subscribe", function() { it("should create a new subscription", async function() { const request = new lib.SubscriptionRequest(RegisteredEvent.name, "subscribe");