RDBC-435 Changes Observable uneven connection state count prematurely closes changes connection

Author: ml054

src/Documents/Changes/ChangesObservable.ts

@@ -59,11 +59,12 @@ export class ChangesObservable<T, TConnectionState extends IChangesConnectionSta
     public off(event: "error", handler: (error: Error) => void);
     public off(event: "data" | "error", handler: ((value: T) => void) | ((error: Error) => void)) {
 
-        this._connectionState.dec();
-
         switch (event) {
             case "data":
-                this._subscribers.delete(handler as (value: T) => void);
+                if (this._subscribers.delete(handler as (value: T) => void)) {
+                    this._connectionState.dec();
+                }
+
                 if (!this._subscribers.size) {
                     // no more subscribers left - remove from parent
                     this._connectionState.removeOnChangeNotification(this._type, this._sendHandler);

test/Ported/Issues/RDBC_435.ts

@@ -0,0 +1,78 @@
+import { testContext, disposeTestDocumentStore } from "../../Utils/TestUtil";
+
+import {
+    IDocumentStore, DocumentChange,
+} from "../../../src";
+import { User } from "../../Assets/Entities";
+import { AsyncQueue } from "../../Utils/AsyncQueue";
+import { throwError } from "../../../src/Exceptions";
+import { assertThat, assertThrows } from "../../Utils/AssertExtensions";
+
+describe("RDBC_435", function () {
+
+    let store: IDocumentStore;
+
+    beforeEach(async function () {
+        store = await testContext.getDocumentStore();
+    });
+
+    afterEach(async () =>
+        await disposeTestDocumentStore(store));
+
+    it("can handle data/errors on/off", async () => {
+
+        const createUser = async () => {
+            const session = store.openSession();
+            const user1 = new User();
+            user1.age = 5;
+            await session.store(user1);
+            await session.saveChanges();
+
+            return user1.id;
+        }
+
+        const changesList = new AsyncQueue<DocumentChange>();
+
+        const changes = store.changes();
+        await changes.ensureConnectedNow();
+
+        const observable = changes.forDocumentsInCollection("users");
+        await observable.ensureSubscribedNow();
+
+        const handler = (change: DocumentChange) => changesList.push(change);
+
+        const errorHandler = e => throwError("InvalidOperationException", e.message);
+
+        observable.on("data", handler);
+        observable.on("error", errorHandler);
+
+        const user1Id = await createUser();
+
+        const change1 = await changesList.poll(15_000);
+        assertThat(change1)
+            .isNotNull();
+        assertThat(change1.id)
+            .isEqualTo(user1Id);
+
+        // now disable error handler, create next user and wait for change
+        observable.off("error", errorHandler);
+
+        const user2Id = await createUser();
+
+        const change2 = await changesList.poll(15_000);
+        assertThat(change2)
+            .isNotNull();
+        assertThat(change2.id)
+            .isEqualTo(user2Id);
+
+        // disable data handle - not we shouldn't get any data
+
+        observable.off("data", handler);
+        await createUser();
+
+        await assertThrows(async () => changesList.poll(3_000), err => {
+            assertThat(err.name)
+                .isEqualTo("TimeoutException");
+        });
+    });
+});