fix(client): Subscription can be disposed only once

Closes enisdenjo#170

src/client.ts

@@ -609,7 +609,6 @@ export function createClient(options: ClientOptions): Client {
             );
 
             releaser = () => {
-              locks--;
               if (!completed && socket.readyState === WebSocketImpl.OPEN)
                 // if not completed already and socket is open, send complete message to server on release
                 socket.send(
@@ -618,6 +617,8 @@ export function createClient(options: ClientOptions): Client {
                     type: MessageType.Complete,
                   }),
                 );
+              locks--;
+              completed = true;
               release();
             };
 
@@ -635,7 +636,10 @@ export function createClient(options: ClientOptions): Client {
         .catch(sink.error) // rejects on close events and errors
         .then(sink.complete); // resolves on release or normal closure
 
-      return () => releaser();
+      return () => {
+        // allow disposing only if not already completed
+        if (!completed) releaser();
+      };
     },
     async dispose() {
       disposed = true;

src/tests/client.ts

@@ -754,6 +754,40 @@ describe('lazy', () => {
       client.close();
     });
   });
+
+  it('should not close connection if a subscription is disposed multiple times', async () => {
+    const { url, ...server } = await startTServer();
+
+    const client = createClient({
+      url,
+      lazy: true,
+      retryAttempts: 0,
+    });
+
+    // create 2 subscriptions
+    const sub0 = tsubscribe(client, {
+      query: 'subscription { ping(key: "0") }',
+    });
+    await server.waitForOperation();
+
+    const sub1 = tsubscribe(client, {
+      query: 'subscription { ping(key: "1") }',
+    });
+    await server.waitForOperation();
+
+    // dispose of the 2nd subscription 2 times (try decrementing locks twice)
+    sub1.dispose();
+    sub1.dispose();
+
+    // first subscription shouldnt complete and the client shouldnt disconnect
+    await sub0.waitForComplete(() => {
+      fail("subscription shouldn't have completed");
+    }, 20);
+
+    await server.waitForClientClose(() => {
+      fail("client shouldn't have closed");
+    }, 20);
+  });
 });
 
 describe('reconnecting', () => {