fix(client): Report close error even if Complete message followed

Closes enisdenjo#245

src/__tests__/client.ts

@@ -12,7 +12,7 @@ import {
   stringifyMessage,
   SubscribePayload,
 } from '../common';
-import { startWSTServer as startTServer } from './utils';
+import { startRawServer, startWSTServer as startTServer } from './utils';
 import { ExecutionResult } from 'graphql';
 
 // silence console.error calls for nicer tests overview
@@ -480,6 +480,65 @@ it('should close socket with error on malformed request', async (done) => {
   );
 });
 
+it('should report close error even if complete message followed', async (done) => {
+  expect.assertions(4);
+
+  const { url, server } = await startRawServer();
+
+  server.on('connection', (socket) => {
+    socket.on('message', (data) => {
+      const msg = parseMessage(String(data));
+
+      // acknowledge conneciton
+      if (msg.type === MessageType.ConnectionInit)
+        socket.send(stringifyMessage({ type: MessageType.ConnectionAck }));
+
+      // respond with a malformed error message and a complete
+      if (msg.type === MessageType.Subscribe) {
+        socket.send(
+          JSON.stringify({
+            id: msg.id,
+            type: MessageType.Error,
+            payload: 'malformed',
+          }),
+        );
+        socket.send(
+          stringifyMessage({ id: msg.id, type: MessageType.Complete }),
+        );
+      }
+    });
+  });
+
+  const client = createClient({
+    url,
+    lazy: false,
+    retryAttempts: 0,
+    onNonLazyError: noop,
+    on: {
+      closed: (err) => {
+        expect((err as CloseEvent).code).toBe(CloseCode.BadRequest);
+        expect((err as CloseEvent).reason).toBe('Invalid message');
+      },
+    },
+  });
+
+  client.subscribe(
+    {
+      query: 'notaquery',
+    },
+    {
+      next: noop,
+      error: (err) => {
+        expect((err as CloseEvent).code).toBe(CloseCode.BadRequest);
+        expect((err as CloseEvent).reason).toBe('Invalid message');
+        client.dispose();
+        done();
+      },
+      complete: noop,
+    },
+  );
+});
+
 describe('ping/pong', () => {
   it('should respond with a pong to a ping', async () => {
     expect.assertions(1);

src/client.ts

@@ -693,6 +693,7 @@ export function createClient(options: ClientOptions): Client {
                 ),
               ]);
             } catch (err) {
+              socket.onmessage = null; // stop reading messages as soon as reading breaks once
               socket.close(
                 CloseCode.BadRequest,
                 err instanceof Error ? err.message : new Error(err).message,