fix(client): Subscribes even if socket is in CLOSING state due to all…

… subscriptions being completed

Closes enisdenjo#173, closes enisdenjo#170

src/client.ts

@@ -459,6 +459,9 @@ export function createClient(options: ClientOptions): Client {
         })(),
       )));
 
+    // if the provided socket is in a closing state, wait for the throw on close
+    if (socket.readyState === WebSocketImpl.CLOSING) await throwOnClose;
+
     let release = () => {
       // releases this connection
     };
@@ -511,12 +514,13 @@ export function createClient(options: ClientOptions): Client {
     )
       throw errOrCloseEvent;
 
-    // disposed or normal closure (completed), shouldnt try again
-    if (
-      disposed ||
-      (isLikeCloseEvent(errOrCloseEvent) && errOrCloseEvent.code === 1000)
-    )
-      return false;
+    // client was disposed, no retries should proceed regardless
+    if (disposed) return false;
+
+    // normal closure (possibly all subscriptions have completed)
+    // if no locks were acquired in the meantime, shouldnt try again
+    if (isLikeCloseEvent(errOrCloseEvent) && errOrCloseEvent.code === 1000)
+      return locks > 0;
 
     // retries are not allowed or we tried to many times, report error
     if (!retryAttempts || retries >= retryAttempts) throw errOrCloseEvent;

src/tests/client.ts

@@ -1183,6 +1183,36 @@ describe('reconnecting', () => {
     // only one retry had happened (for first subscription)
     expect(retry).toBeCalledTimes(1);
   });
+
+  it('should subscribe even if socket is in CLOSING state due to all subscriptions being completed', async () => {
+    const { url, ...server } = await startTServer();
+
+    const client = createClient({
+      url,
+      lazy: true,
+      retryAttempts: 0,
+    });
+
+    // subscribe and wait for operation
+    let sub = tsubscribe(client, {
+      query: 'subscription { ping }',
+    });
+    await server.waitForOperation();
+
+    // complete the subscription
+    sub.dispose();
+
+    // give room for the socket close in the stack
+    await new Promise((resolve) => setImmediate(resolve));
+
+    // immediately subscribe again
+    sub = tsubscribe(client, {
+      query: 'subscription { ping }',
+    });
+
+    // the new subscription should go through
+    await server.waitForOperation();
+  });
 });
 
 describe('events', () => {