From 15cd35301e8810822af95e30be50759fdccabe8e Mon Sep 17 00:00:00 2001 From: Ekhorn Date: Thu, 5 Nov 2026 17:24:12 +0100 Subject: [PATCH 1/2] Fix event subscriptions stopping after a reconnect A socket error or close during a subscribe left the pending reply state set, so every later subscribe was rejected before anything was sent. Events only came back after a page reload. - Clear the pending subscribe reply on an unexpected disconnect - Stop reusing a rejected shared asset or attribute subscription, and remove the callback of the failed call - Detach and close the previous socket before opening a new one - Restore subscriptions after a reconnect - Let unsubscribe cancel a subscription that is still being restored - Add component tests and npmTest for the core package --- ui/component/core/build.gradle | 5 + ui/component/core/package.json | 2 +- ui/component/core/playwright.config.ts | 21 ++ ui/component/core/src/event.ts | 70 +++++- ui/component/core/test/event-provider.test.ts | 205 ++++++++++++++++++ .../test/fixtures/event-provider-harness.js | 197 +++++++++++++++++ 6 files changed, 488 insertions(+), 12 deletions(-) create mode 100644 ui/component/core/playwright.config.ts create mode 100644 ui/component/core/test/event-provider.test.ts create mode 100644 ui/component/core/test/fixtures/event-provider-harness.js diff --git a/ui/component/core/build.gradle b/ui/component/core/build.gradle index ddfbc0fd32..fdface71b8 100644 --- a/ui/component/core/build.gradle +++ b/ui/component/core/build.gradle @@ -9,3 +9,8 @@ tasks.register('prepareUi') { tasks.register('publishUi') { dependsOn clean, npmPublish } + +tasks.register('npmTest', Exec) { + dependsOn getYarnInstallTask() + commandLine npmCommand("yarn"), "run", "test" +} diff --git a/ui/component/core/package.json b/ui/component/core/package.json index cc9541a3d4..11fcf99c77 100644 --- a/ui/component/core/package.json +++ b/ui/component/core/package.json @@ -16,7 +16,7 @@ ], "scripts": { "build": "npx tsc -b", - "test": "echo \"No tests\" && exit 0", + "test": "npx tsc -b && npx playwright test", "prepack": "npx rspack" }, "author": "OpenRemote", diff --git a/ui/component/core/playwright.config.ts b/ui/component/core/playwright.config.ts new file mode 100644 index 0000000000..5fdc4a35d1 --- /dev/null +++ b/ui/component/core/playwright.config.ts @@ -0,0 +1,21 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { defineCtConfig } from "@openremote/test/component.config"; + +export default defineCtConfig(__dirname); diff --git a/ui/component/core/src/event.ts b/ui/component/core/src/event.ts index e59ead4b3f..ed6154985b 100644 --- a/ui/component/core/src/event.ts +++ b/ui/component/core/src/event.ts @@ -90,6 +90,8 @@ interface EventSubscriptionInfo { eventSubscription: EventSubscription; callback: (event: T) => void; deferred: Deferred | null; + // Unsubscribed while the server has yet to confirm it + cancelled?: boolean; } interface AssetSubscriptionInfo { @@ -229,8 +231,12 @@ abstract class EventProviderImpl implements EventProvider { const subscription = this._pendingSubscription; this._pendingSubscription = null; - // Store subscriptionId and callback - this._subscriptionMap.set(subscriptionId, subscription); + if (subscription.cancelled) { + this._doUnsubscribe(subscriptionId); + } else { + // Store subscriptionId and callback + this._subscriptionMap.set(subscriptionId, subscription); + } this._processNextSubscription(); const deferred = subscription.deferred; @@ -266,6 +272,20 @@ abstract class EventProviderImpl implements EventProvider { return; } + // Restored subscriptions are not in the map until the server has confirmed them again + const queuedIndex = this._queuedSubscriptions.findIndex( + (info) => info.eventSubscription.subscriptionId === subscriptionId + ); + if (queuedIndex !== -1) { + this._queuedSubscriptions.splice(queuedIndex, 1); + return; + } + + if (this._pendingSubscription?.eventSubscription.subscriptionId === subscriptionId) { + this._pendingSubscription.cancelled = true; + return; + } + if (this._assetEventCallbackMap.delete(subscriptionId)) { if (this._assetEventPromise && this._assetEventCallbackMap.size === 0) { // Remove the asset subscription @@ -330,8 +350,12 @@ abstract class EventProviderImpl implements EventProvider { // assetIds: assetIds // }; // } + // A failed subscription must not be reused this._assetEventPromise = this.subscribe(subscription, (event) => { this._assetEventCallbackMap.forEach((callback) => callback(event)); + }).catch((reason) => { + this._assetEventPromise = undefined; + throw reason; }); } @@ -352,6 +376,9 @@ abstract class EventProviderImpl implements EventProvider { this._assetEventCallbackMap.set(subscriptionId, eventFilter); + // The caller doesn't get an ID to unsubscribe with when this fails + this._assetEventPromise.catch(() => this._assetEventCallbackMap.delete(subscriptionId)); + return this._assetEventPromise.then(() => { try { // Get the current state of the assets if requested @@ -418,8 +445,12 @@ abstract class EventProviderImpl implements EventProvider { eventType: "attribute", filter: assetFilter, }; + // A failed subscription must not be reused this._attributeEventPromise = this.subscribe(subscription, (event) => { this._attributeEventCallbackMap.forEach((callback) => callback(event)); + }).catch((reason) => { + this._attributeEventPromise = undefined; + throw reason; }); } @@ -447,6 +478,9 @@ abstract class EventProviderImpl implements EventProvider { this._attributeEventCallbackMap.set(subscriptionId, eventFilter); + // The caller doesn't get an ID to unsubscribe with when this fails + this._attributeEventPromise.catch(() => this._attributeEventCallbackMap.delete(subscriptionId)); + return this._attributeEventPromise.then(() => { try { // Get the current state of the attributes if requested @@ -551,14 +585,8 @@ abstract class EventProviderImpl implements EventProvider { } protected _onConnect() { - if (Object.keys(this._subscriptionMap).length > 0) { - for (const subscriptionId in this._subscriptionMap) { - if (this._subscriptionMap.has(subscriptionId)) { - this._queuedSubscriptions.unshift(this._subscriptionMap.get(subscriptionId)!); - } - } - this._subscriptionMap.clear(); - } + this._subscriptionMap.forEach((subscription) => this._queuedSubscriptions.unshift(subscription)); + this._subscriptionMap.clear(); this._processNextSubscription(); } @@ -568,7 +596,9 @@ abstract class EventProviderImpl implements EventProvider { this._onStatusChanged(EventProviderStatus.DISCONNECTED); } if (this._pendingSubscription) { - this._queuedSubscriptions.unshift(this._pendingSubscription); + if (!this._pendingSubscription.cancelled) { + this._queuedSubscriptions.unshift(this._pendingSubscription); + } this._pendingSubscription = null; } @@ -646,6 +676,8 @@ export class WebSocketEventProvider extends EventProviderImpl { } protected _doConnect(): Promise { + this._discardWebSocket(); + let authorisedUrl = this._endpointUrl + "?Realm=" + manager.config.realm; this._connectDeferred = new Deferred(); @@ -741,7 +773,23 @@ export class WebSocketEventProvider extends EventProviderImpl { return this._connectDeferred.promise; } + // Only the current socket may change the state of the provider + protected _discardWebSocket(): void { + if (!this._webSocket) { + return; + } + + this._webSocket.onopen = null; + this._webSocket.onerror = null; + this._webSocket.onclose = null; + this._webSocket.onmessage = null; + this._webSocket.close(); + this._webSocket = undefined; + } + protected _beforeDisconnect(): void { + // No reply can arrive anymore and the pending subscription is sent again after reconnecting + this._subscribeDeferred = null; this._onDisconnect(); } diff --git a/ui/component/core/test/event-provider.test.ts b/ui/component/core/test/event-provider.test.ts new file mode 100644 index 0000000000..df8427c4eb --- /dev/null +++ b/ui/component/core/test/event-provider.test.ts @@ -0,0 +1,205 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { ct, expect } from "@openremote/test"; + +import { EventProviderHarness } from "./fixtures/event-provider-harness.js"; + +ct("should send a subscribe again when the connection drops before the reply", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + el.subscribeReply = "none"; + const interrupted = el.subscribeTo("alarm", "interrupted"); + await el.settle(); + + el.subscribeReply = "subscribed"; + await el.reconnect(); + const interruptedOutcome = await el.outcome(interrupted); + const nextOutcome = await el.outcome(el.subscribeTo("alarm", "next")); + + return { interruptedOutcome, nextOutcome, frames: el.socket.frames }; + }); + + // The slot for the reply used to stay taken after a drop, so nothing went out on the new socket and every later + // subscribe was rejected with "There's already a pending subscription" + expect(result.frames).toEqual(["SUBSCRIBE", "SUBSCRIBE"]); + expect(result.interruptedOutcome).toBe("resolved"); + expect(result.nextOutcome).toBe("resolved"); +}); + +for (const eventType of ["asset", "attribute"]) { + ct(`should retry a shared ${eventType} subscription after it was rejected`, async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness, type: string) => { + await el.connect(); + el.subscribeReply = "unauthorized"; + const rejectedOutcome = await el.outcome(el.subscribeTo(type, "rejected")); + + el.subscribeReply = "subscribed"; + const retriedOutcome = await el.outcome(el.subscribeTo(type, "retried")); + el.trigger(type); + + return { rejectedOutcome, retriedOutcome, frames: el.socket.frames, received: el.received }; + }, eventType); + + expect(result.rejectedOutcome).toBe("rejected: Unauthorized"); + // The rejected promise used to be kept and shared, so a retry got the same rejection without sending anything + expect(result.retriedOutcome).toBe("resolved"); + expect(result.frames).toEqual(["SUBSCRIBE", "SUBSCRIBE"]); + // The callback of the failed call used to stay registered, and nothing could remove it + expect(result.received).toEqual(["retried"]); + }); +} + +ct("should ignore a replaced socket that closes during the reconnect", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + const replaced = el.socket; + replaced.error(); + await el.until(() => el.socket !== replaced); + const current = el.socket; + + replaced.closed(); + current.open(); + await el.settle(); + + return { status: el.provider.status, closeRequested: replaced.closeRequested, socketCount: el.sockets.length }; + }); + + // The late close used to fail the reconnect that was under way, which left the new socket open but unused + expect(result.status).toBe("CONNECTED"); + expect(result.closeRequested).toBe(true); + expect(result.socketCount).toBe(2); +}); + +ct("should keep the connection when a replaced socket closes during a subscribe", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + const replaced = el.socket; + replaced.error(); + await el.until(() => el.socket !== replaced); + const current = el.socket; + current.open(); + await el.until(() => el.provider.status === "CONNECTED"); + + el.subscribeReply = "none"; + const subscribed = el.subscribeTo("alarm", "alarm"); + await el.settle(); + replaced.closed(); + await el.settle(); + current.answer(); + + return { outcome: await el.outcome(subscribed), status: el.provider.status, socketCount: el.sockets.length }; + }); + + // The close used to count as a drop of the current socket, which started another reconnect and left that socket open + expect(result.status).toBe("CONNECTED"); + expect(result.socketCount).toBe(2); + expect(result.outcome).toBe("resolved"); +}); + +for (const eventType of ["alarm", "asset", "attribute"]) { + ct(`should keep delivering ${eventType} events after a reconnect`, async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness, type: string) => { + await el.connect(); + await el.subscribeTo(type, "subscribed"); + await el.reconnect(); + el.trigger(type); + + return { frames: el.socket.frames, received: el.received }; + }, eventType); + + // Subscriptions used to be lost on a reconnect, so a consumer holding on to its ID silently stopped getting events + expect(result.frames).toEqual(["SUBSCRIBE"]); + expect(result.received).toEqual(["subscribed"]); + }); +} + +ct("should end with one server subscription when resubscribing during a restore", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + const subscriptionId = await el.subscribeTo("attribute", "old"); + + el.subscribeReply = "none"; + await el.reconnect(); + await el.provider.unsubscribe(subscriptionId); + el.subscribeReply = "subscribed"; + const resubscribed = el.subscribeTo("attribute", "new"); + el.socket.answer(); + const outcome = await el.outcome(resubscribed); + el.trigger("attribute"); + + return { outcome, frames: el.socket.frames, subscriptions: el.socket.subscriptions.size, received: el.received }; + }); + + expect(result.outcome).toBe("resolved"); + expect(result.frames).toEqual(["SUBSCRIBE", "UNSUBSCRIBE", "SUBSCRIBE"]); + expect(result.subscriptions).toBe(1); + expect(result.received).toEqual(["new"]); +}); + +ct("should not restore a subscription unsubscribed while it waits for its turn", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + const subscriptionIds = [await el.subscribeTo("alarm", "first"), await el.subscribeTo("alarm", "second")]; + + el.subscribeReply = "none"; + await el.reconnect(); + const inFlight = el.socket.unanswered[0].subscriptionId; + const waiting = subscriptionIds.find((subscriptionId) => subscriptionId !== inFlight); + await el.provider.unsubscribe(waiting); + el.socket.answer(); + await el.settle(); + + return { inFlight, frames: el.socket.frames, subscriptions: [...el.socket.subscriptions.keys()] }; + }); + + expect(result.frames).toEqual(["SUBSCRIBE"]); + expect(result.subscriptions).toEqual([result.inFlight]); +}); + +ct("should not restore a subscription unsubscribed before another drop", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + const subscriptionId = await el.subscribeTo("alarm", "alarm"); + + el.subscribeReply = "none"; + await el.reconnect(); + await el.provider.unsubscribe(subscriptionId); + await el.reconnect(); + + return { frames: el.socket.frames }; + }); + + expect(result.frames).toEqual([]); +}); diff --git a/ui/component/core/test/fixtures/event-provider-harness.js b/ui/component/core/test/fixtures/event-provider-harness.js new file mode 100644 index 0000000000..a64cbb7ffd --- /dev/null +++ b/ui/component/core/test/fixtures/event-provider-harness.js @@ -0,0 +1,197 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +// Plain JS on purpose: the component-test bundle compiles .ts files against ui/test's tsconfig +// (rootDir ui/test), so TS component sources outside that dir fail the build. +import { WebSocketEventProvider } from "@openremote/core"; + +const SETTLE_MILLIS = 50; +const WAIT_MILLIS = 2000; + +const ASSET_ID = "asset1"; + +const EVENTS = { + alarm: { eventType: "alarm" }, + asset: { eventType: "asset", asset: { id: ASSET_ID } }, + attribute: { eventType: "attribute", ref: { id: ASSET_ID, name: "attribute1" } }, +}; + +const sleep = (millis) => new Promise((resolve) => setTimeout(resolve, millis)); + +/** + * Stands in for a browser socket and the server behind it, which holds the subscriptions of that connection. The + * socket's own events only fire when a test calls `open`, `error` or `closed`, so a test decides their order. + */ +class FakeWebSocket { + static harness; + + constructor() { + this.frames = []; + this.subscriptions = new Map(); + this.unanswered = []; + this.closeRequested = false; + FakeWebSocket.harness.sockets.push(this); + } + + send(message) { + const separator = message.indexOf(":"); + const frame = message.substring(0, separator); + const body = JSON.parse(message.substring(separator + 1)); + this.frames.push(frame); + + if (frame === "UNSUBSCRIBE") { + this.subscriptions.delete(body.subscriptionId); + } else if (frame === "SUBSCRIBE") { + this._onSubscribe(body); + } + } + + _onSubscribe(subscription) { + const reply = FakeWebSocket.harness.subscribeReply; + + if (reply === "none") { + this.unanswered.push(subscription); + } else if (reply === "unauthorized") { + setTimeout(() => this.onmessage?.({ data: "UNAUTHORIZED:" + JSON.stringify(subscription) })); + } else { + this.subscriptions.set(subscription.subscriptionId, subscription); + setTimeout(() => this.onmessage?.({ data: "SUBSCRIBED:" + JSON.stringify(subscription) })); + } + } + + /** Replies to the subscribes that got no reply so far. */ + answer() { + const subscriptions = this.unanswered; + this.unanswered = []; + subscriptions.forEach((subscription) => { + this.subscriptions.set(subscription.subscriptionId, subscription); + this.onmessage?.({ data: "SUBSCRIBED:" + JSON.stringify(subscription) }); + }); + } + + close() { + this.closeRequested = true; + } + + open() { + this.onopen?.({}); + } + + error() { + this.onerror?.({}); + } + + closed() { + this.onclose?.({}); + } +} + +/** + * Runs a real {@link WebSocketEventProvider} against {@link FakeWebSocket}. Tests drive it from `evaluate`, since the + * provider and its callbacks can't cross into the Playwright test. + */ +export class EventProviderHarness extends HTMLElement { + constructor() { + super(); + this.sockets = []; + /** How the server replies to a subscribe: `subscribed`, `unauthorized` or `none`. */ + this.subscribeReply = "subscribed"; + /** The labels of the subscribe calls whose callback got an event, in order. */ + this.received = []; + + FakeWebSocket.harness = this; + window.WebSocket = FakeWebSocket; + this.provider = new WebSocketEventProvider("https://localhost"); + } + + /** The socket the provider opened last. */ + get socket() { + return this.sockets[this.sockets.length - 1]; + } + + /** Subscribes through the method a consumer of this event type uses, and records events under `label`. */ + subscribeTo(eventType, label) { + const callback = () => this.received.push(label); + + switch (eventType) { + case "asset": + return this.provider.subscribeAssetEvents([ASSET_ID], false, callback); + case "attribute": + return this.provider.subscribeAttributeEvents([ASSET_ID], false, callback); + default: + return this.provider.subscribe({ eventType }, callback); + } + } + + /** Sends an event of this type to each subscription of that type on the current connection. */ + trigger(eventType) { + this.socket.subscriptions.forEach((subscription, subscriptionId) => { + if (subscription.eventType === eventType) { + const triggered = { subscriptionId, events: [EVENTS[eventType]] }; + this.socket.onmessage?.({ data: "TRIGGERED:" + JSON.stringify(triggered) }); + } + }); + } + + async connect() { + this.provider.connect(); + await this.until(() => this.socket); + this.socket.open(); + await this.until(() => this.provider.status === "CONNECTED"); + await this.settle(); + } + + /** Drops the current connection and completes the reconnect the provider starts by itself. */ + async reconnect() { + const previous = this.socket; + previous.error(); + previous.closed(); + await this.until(() => this.socket !== previous); + this.socket.open(); + await this.until(() => this.provider.status === "CONNECTED"); + await this.settle(); + } + + async until(predicate) { + const deadline = Date.now() + WAIT_MILLIS; + while (!predicate()) { + if (Date.now() > deadline) { + throw new Error("Timed out waiting for: " + predicate); + } + await sleep(1); + } + } + + /** Gives the provider time to finish what it does on its own, to check that something didn't happen. */ + settle() { + return sleep(SETTLE_MILLIS); + } + + /** Describes how a promise ended, so a hanging one fails an assertion instead of the test's timeout. */ + outcome(promise) { + return Promise.race([ + promise.then( + () => "resolved", + (reason) => "rejected: " + reason + ), + sleep(WAIT_MILLIS).then(() => "pending"), + ]); + } +} + +customElements.define("event-provider-harness", EventProviderHarness); From 5f4aff2c5b0d536739cd0d662a1d79b61ec4c2f2 Mon Sep 17 00:00:00 2001 From: Ekhorn Date: Fri, 6 Nov 2026 14:50:30 +0100 Subject: [PATCH 2/2] Resolve the subscribe call of a subscription unsubscribed early Unsubscribing a subscription the server hadn't confirmed yet left its subscribe promise pending. It now resolves with the ID, as it already did when the reply arrived after the unsubscribe. --- ui/component/core/src/event.ts | 19 +++++++--- ui/component/core/test/event-provider.test.ts | 36 +++++++++++++++++++ 2 files changed, 51 insertions(+), 4 deletions(-) diff --git a/ui/component/core/src/event.ts b/ui/component/core/src/event.ts index ed6154985b..d0cfcd1cd2 100644 --- a/ui/component/core/src/event.ts +++ b/ui/component/core/src/event.ts @@ -277,7 +277,8 @@ abstract class EventProviderImpl implements EventProvider { (info) => info.eventSubscription.subscriptionId === subscriptionId ); if (queuedIndex !== -1) { - this._queuedSubscriptions.splice(queuedIndex, 1); + const [queued] = this._queuedSubscriptions.splice(queuedIndex, 1); + this._resolveUnsubscribed(queued); return; } @@ -596,16 +597,26 @@ abstract class EventProviderImpl implements EventProvider { this._onStatusChanged(EventProviderStatus.DISCONNECTED); } if (this._pendingSubscription) { - if (!this._pendingSubscription.cancelled) { - this._queuedSubscriptions.unshift(this._pendingSubscription); - } + const pending = this._pendingSubscription; this._pendingSubscription = null; + + if (pending.cancelled) { + this._resolveUnsubscribed(pending); + } else { + this._queuedSubscriptions.unshift(pending); + } } this._onStatusChanged(EventProviderStatus.CONNECTING); this._scheduleReconnect(); } + // Resolves rather than rejects, since the subscribe didn't fail but was unsubscribed + protected _resolveUnsubscribed(subscription: EventSubscriptionInfo) { + subscription.deferred?.resolve(subscription.eventSubscription.subscriptionId!); + subscription.deferred = null; + } + protected _scheduleReconnect() { if (this._reconnectTimer) { return; diff --git a/ui/component/core/test/event-provider.test.ts b/ui/component/core/test/event-provider.test.ts index df8427c4eb..5799b18f25 100644 --- a/ui/component/core/test/event-provider.test.ts +++ b/ui/component/core/test/event-provider.test.ts @@ -203,3 +203,39 @@ ct("should not restore a subscription unsubscribed before another drop", async ( expect(result.frames).toEqual([]); }); + +ct("should resolve a subscribe that was unsubscribed before it was sent", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + const subscribed = el.provider.subscribe({ eventType: "alarm", subscriptionId: "own" }, () => {}); + await el.provider.unsubscribe("own"); + const outcome = await el.outcome(subscribed); + await el.connect(); + + return { outcome, frames: el.socket.frames }; + }); + + expect(result.outcome).toBe("resolved"); + expect(result.frames).toEqual([]); +}); + +ct("should resolve a subscribe unsubscribed before its reply when the connection drops", async ({ mount }) => { + const component = await mount(EventProviderHarness); + + const result = await component.evaluate(async (el: EventProviderHarness) => { + await el.connect(); + el.subscribeReply = "none"; + const subscribed = el.provider.subscribe({ eventType: "alarm", subscriptionId: "own" }, () => {}); + await el.settle(); + await el.provider.unsubscribe("own"); + + el.subscribeReply = "subscribed"; + await el.reconnect(); + + return { outcome: await el.outcome(subscribed), frames: el.socket.frames }; + }); + + expect(result.outcome).toBe("resolved"); + expect(result.frames).toEqual([]); +});