Merge pull request #1058 from bbassie/fix/subscriber-account-update

Resend subscriptions when the account permissions change

packages/lib/src/subscriptions.ts

@@ -473,6 +473,8 @@ export class EventSubscriber<E, S = null> {
 	private _callbacks = new Array<EventSubscriberCallback<E>>();
 	/** Filters applied to the filtered handler, if empty the filtered handler will not be called */
 	private _filters = SubscriptionFilters.empty();
+	/** True if a resend of the subscription is queued for the current tick */
+	private _resendQueued = false;
 
 	constructor(
 		protected Event: EventClass<E>,
@@ -488,6 +490,13 @@ export class EventSubscriber<E, S = null> {
 		webSocketConnector.on("close", () => {
 			this.handleConnectionEvent("close");
 		});
+		// Only the web Control reports account changes
+		const accountControl = control as Link & { onAccountUpdate?: (handler: () => void) => void };
+		if (typeof accountControl.onAccountUpdate === "function") {
+			accountControl.onAccountUpdate(() => {
+				this.handleAccountEvent();
+			});
+		}
 	}
 
 	/**
@@ -497,13 +506,38 @@ export class EventSubscriber<E, S = null> {
 	 */
 	handleConnectionEvent(event: "connect" | "drop" | "resume" | "close") {
 		if (event === "connect" || event === "resume") {
-			this._sendRequest("replace", this._callbacks.length > 0 ? SubscriptionFilters.all() : this._filters);
+			this._queueResend();
 		} else if (this.synced) {
 			this.synced = false;
 			this._notify(null);
 		}
 	}
 
+	/**
+	 * Handle the permissions of the account changing
+	 * Resends the subscription so it fails if permission was lost, or syncs if it was granted.
+	 */
+	handleAccountEvent() {
+		if (this._hasSubscriptions()) {
+			this._queueResend();
+		}
+	}
+
+	/**
+	 * Resend the current subscription at the end of the tick, the web Control
+	 * reports the account on connect so both would otherwise send it.
+	 */
+	private _queueResend() {
+		if (this._resendQueued) {
+			return;
+		}
+		this._resendQueued = true;
+		queueMicrotask(() => {
+			this._resendQueued = false;
+			this._sendRequest("replace", this._callbacks.length > 0 ? SubscriptionFilters.all() : this._filters);
+		});
+	}
+
 	/**
 	 * Handle incoming events and distribute it to the correct callbacks
 	 * @param event - event from subscribed resource
@@ -657,6 +691,8 @@ export class EventSubscriber<E, S = null> {
 			if (err instanceof Error) {
 				this.error = err;
 			}
+			this.synced = false;
+			this._notify(null);
 		}
 	}
 }

test/lib/subscriptions.test.js

@@ -1232,6 +1232,66 @@ describe("lib/subscriptions", function() {
 			});
 		});
 
+		describe(".handleAccountEvent()", function() {
+			it("should send replace(all) if subscribed", async function() {
+				eventSubscriber.subscribe(() => {});
+				await onceConnectorSend();
+
+				eventSubscriber.handleAccountEvent();
+				await onceConnectorSend();
+				assertLastMessage(new lib.SubscriptionRequest(
+					RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all()
+				));
+			});
+
+			it("should send replace(filters) when only filters exist", async function() {
+				eventSubscriber.addFilters("foo");
+				await onceConnectorSend();
+
+				eventSubscriber.handleAccountEvent();
+				await onceConnectorSend();
+				assertLastMessage(new lib.SubscriptionRequest(
+					RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.fromShorthand("foo")
+				));
+			});
+
+			it("should send nothing if not subscribed", async function() {
+				eventSubscriber.handleAccountEvent();
+				await new Promise(r => setImmediate(r));
+				assertNoMessageSent();
+			});
+
+			it("should send once when connecting and the account is reported", async function() {
+				eventSubscriber.subscribe(() => {});
+				await onceConnectorSend();
+				mockControl.connector.sentMessages.pop();
+
+				eventSubscriber.handleAccountEvent();
+				mockControl.connector.emit("connect");
+				await new Promise(r => setImmediate(r));
+				assert.equal(mockControl.connector.sentMessages.length, 1);
+			});
+
+			it("should be called on account updates from the control", async function() {
+				const handlers = [];
+				mockControl = new MockControl(new MockConnector(
+					addr({ controlId: 0 }),
+					addr("controller"),
+				));
+				mockControl.onAccountUpdate = handler => { handlers.push(handler); };
+				const subscriber = new lib.EventSubscriber(RegisteredEvent, mockControl);
+				assert.equal(handlers.length, 1);
+
+				subscriber.subscribe(() => {});
+				await onceConnectorSend();
+				handlers[0]({ name: "test", roles: [] });
+				await onceConnectorSend();
+				assertLastMessage(new lib.SubscriptionRequest(
+					RegisteredEvent.name, "replace", -1, lib.SubscriptionFilters.all()
+				));
+			});
+		});
+
 		describe("._sendRequest()", function() {
 			it("should do nothing if connection is not valid", async function() {
 				mockControl.connector.valid = false;
@@ -1313,6 +1373,26 @@ describe("lib/subscriptions", function() {
 				assert.equal(logged, undefined);
 			});
 
+			it("should notify callbacks as not synced when the request fails", async function() {
+				let calledWith;
+				eventSubscriber.subscribe((e, s) => { calledWith = [e, s]; });
+				const [msg] = await onceConnectorSend();
+				mockControl.connector.emit("message", new lib.MessageResponse(0, msg.dst, msg.src,
+					lib.SubscriptionRequest.Response.fromJSON(false)
+				));
+				await new Promise(r => setImmediate(r));
+				assert.equal(eventSubscriber.synced, true);
+
+				mockControl.sendTo = async function() {
+					throw new lib.PermissionError("Permission denied");
+				};
+				eventSubscriber.handleAccountEvent();
+				await new Promise(r => setImmediate(r));
+				assert.equal(eventSubscriber.synced, false);
+				assert.deepEqual(calledWith, [null, false]);
+				assert.ok(eventSubscriber.error instanceof lib.PermissionError);
+			});
+
 			it("should expose errors through the snapshot", async function() {
 				eventSubscriber.control.sendTo = function() {
 					throw new lib.PermissionError("Permission denied");