Merge pull request #958 from bbassie/feature/subscription-event-address

Stop addressing subscription events to a request id

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

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;
 			}
 		}

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));

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");