diff --git a/packages/lib/src/subscriptions.ts b/packages/lib/src/subscriptions.ts index c6279375..8aa03c38 100644 --- a/packages/lib/src/subscriptions.ts +++ b/packages/lib/src/subscriptions.ts @@ -473,6 +473,8 @@ export class EventSubscriber { private _callbacks = new Array>(); /** 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, @@ -488,6 +490,13 @@ export class EventSubscriber { 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 { */ 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 { if (err instanceof Error) { this.error = err; } + this.synced = false; + this._notify(null); } } } diff --git a/test/lib/subscriptions.test.js b/test/lib/subscriptions.test.js index 585549f4..46c4c7e0 100644 --- a/test/lib/subscriptions.test.js +++ b/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");