Repository: mcollina/fastify-gql
Pull request #270 (2020-08-31 09:48:41)

fix: handle SubscriptionClient ready state properly
This PR fixes the ready state of the `SubscriptionClient` that was not properly handled due to an invalid property set in the `close` function (`this.isReady = false` instead of `this.ready = false`). 
It also extends the `connectionCallback` usage to handle error payload in case of a `connection_error`.

PacoDu: It also throws an error if createSubscription is called and the connection is not ready, otherwise the gateway was accepting gql_start without any error even if the connection_init failed with the related service.
PacoDu: It is called when a new connection is established and I thought it was a good fit to also handle the error case like a 'classic' callback with the error defined if it failed, but this can be a breaking change for some users if not handled. 
If you prefer I can declare a new callback but I think it makes the interface more complicated and it's not necessary.
PacoDu: I've added the `failedConnectionCallback` to prevent a breaking change on `connectionCallback`.

Patch:
diff --git a/README.md b/README.md
index 2702238..7b4ec7a 100644
--- a/README.md
+++ b/README.md
@@ -904,6 +904,7 @@ __fastify-gql__ supports the following options:
       * `wsConnectionParams.reconnect`: `Boolean` Enable reconnect on connection close (Default: `false`)
       * `wsConnectionParams.maxReconnectAttempts`: `Number` Defines the maximum reconnect attempts if reconnect is enabled (Default: `Infinity`)
       * `wsConnectionParams.connectionCallback`: `Function` A function called after a `connection_ack` message is received.
+      * `wsConnectionParams.failedConnectionCallback`: `Function` A function called after a `connection_error` message is received, the first argument contains the message payload.
       * `wsConnectionParams.failedReconnectCallback`: `Function` A function called if reconnect is enabled and maxReconnectAttempts is reached.
 
 * `persistedQueries`: A hash/query map to resolve the full query text using it's unique hash. Overrides `persistedQueryProvider`.
diff --git a/lib/subscription-client.js b/lib/subscription-client.js
index 05580f2..6558279 100644
--- a/lib/subscription-client.js
+++ b/lib/subscription-client.js
@@ -28,6 +28,7 @@ class SubscriptionClient {
       maxReconnectAttempts = Infinity,
       serviceName,
       connectionCallback,
+      failedConnectionCallback,
       failedReconnectCallback,
       connectionInitPayload
     } = config
@@ -38,6 +39,7 @@ class SubscriptionClient {
     this.serviceName = serviceName
     this.reconnectAttempts = 0
     this.connectionCallback = connectionCallback
+    this.failedConnectionCallback = failedConnectionCallback
     this.failedReconnectCallback = failedReconnectCallback
     this.connectionInitPayload = connectionInitPayload
 
@@ -68,14 +70,14 @@ class SubscriptionClient {
 
     this.socket.onerror = () => {}
 
-    this.socket.onmessage = ({ data }) => {
-      this.handleMessage(data)
+    this.socket.onmessage = async ({ data }) => {
+      await this.handleMessage(data)
     }
   }
 
   close (tryReconnect, closedByUser = true) {
     this.closedByUser = closedByUser
-    this.isReady = false
+    this.ready = false
 
     if (this.socket !== null) {
       if (closedByUser) {
@@ -154,7 +156,7 @@ class SubscriptionClient {
     )
   }
 
-  handleMessage (message) {
+  async handleMessage (message) {
     let data
     let operationId
     let operation
@@ -204,6 +206,9 @@ class SubscriptionClient {
         break
       case GQL_CONNECTION_ERROR:
         this.close(this.tryReconnect, false)
+        if (this.failedConnectionCallback) {
+          await this.failedConnectionCallback(data.payload)
+        }
         break
       case GQL_CONNECTION_KEEP_ALIVE:
         break
@@ -216,7 +221,10 @@ class SubscriptionClient {
 
   startOperation (operationId) {
     const { started, options, handler } = this.operations.get(operationId)
-    if (this.ready && !started) {
+    if (!started) {
+      if (!this.ready) {
+        throw new Error('Connection is not ready')
+      }
       this.operations.set(operationId, { started: true, options, handler })
       this.sendMessage(operationId, GQL_START, options)
     }


Test patch:
diff --git a/test/subscription-client.js b/test/subscription-client.js
index f805606..ceab7de 100644
--- a/test/subscription-client.js
+++ b/test/subscription-client.js
@@ -13,30 +13,31 @@ test('subscription client calls the publish method with the correct payload', (t
   server.on('connection', function connection (ws) {
     ws.on('message', function incoming (message) {
       const data = JSON.parse(message)
-      if (data.type === 'start') {
+      if (data.type === 'connection_init') {
+        ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
+      } else if (data.type === 'start') {
         ws.send(JSON.stringify({ id: '1', type: 'data', payload: { data: { foo: 'bar' } } }))
       }
     })
-
-    ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
   })
 
   const client = new SubscriptionClient(`ws://localhost:${port}`, {
     reconnect: true,
     maxReconnectAttempts: 10,
-    serviceName: 'test-service'
-  })
-
-  client.createSubscription('query', {}, (data) => {
-    t.deepEqual(data, {
-      topic: 'test-service_1',
-      payload: {
-        foo: 'bar'
-      }
-    })
-    client.close()
-    server.close()
-    t.end()
+    serviceName: 'test-service',
+    connectionCallback: () => {
+      client.createSubscription('query', {}, (data) => {
+        t.deepEqual(data, {
+          topic: 'test-service_1',
+          payload: {
+            foo: 'bar'
+          }
+        })
+        client.close()
+        server.close()
+        t.end()
+      })
+    }
   })
 })
 
@@ -47,28 +48,29 @@ test('subscription client calls the publish method with null after GQL_COMPLETE
   server.on('connection', function connection (ws) {
     ws.on('message', function incoming (message) {
       const data = JSON.parse(message)
-      if (data.type === 'start') {
+      if (data.type === 'connection_init') {
+        ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
+      } else if (data.type === 'start') {
         ws.send(JSON.stringify({ id: '1', type: 'complete' }))
       }
     })
-
-    ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
   })
 
   const client = new SubscriptionClient(`ws://localhost:${port}`, {
     reconnect: true,
     maxReconnectAttempts: 10,
-    serviceName: 'test-service'
-  })
-
-  client.createSubscription('query', {}, (data) => {
-    t.deepEqual(data, {
-      topic: 'test-service_1',
-      payload: null
-    })
-    client.close()
-    server.close()
-    t.end()
+    serviceName: 'test-service',
+    connectionCallback: () => {
+      client.createSubscription('query', {}, (data) => {
+        t.deepEqual(data, {
+          topic: 'test-service_1',
+          payload: null
+        })
+        client.close()
+        server.close()
+        t.end()
+      })
+    }
   })
 })
 
@@ -141,13 +143,6 @@ test('subscription client stops trying reconnecting after maxReconnectAttempts',
     }
   })
   server.close()
-
-  client.createSubscription('query', {}, (data) => {
-    t.deepEqual(data, {
-      topic: 'test-service_1',
-      payload: null
-    })
-  })
 })
 
 test('subscription client multiple subscriptions is handled by one operation', t => {
@@ -157,18 +152,22 @@ test('subscription client multiple subscriptions is handled by one operation', t
   server.on('connection', function connection (ws) {
     ws.on('message', function incoming (message) {
       const data = JSON.parse(message)
-      if (data.type === 'start') {
+      if (data.type === 'connection_init') {
+        ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
+      } else if (data.type === 'start') {
         ws.send(JSON.stringify({ id: '1', type: 'complete' }))
       }
     })
-
-    ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
   })
 
   const client = new SubscriptionClient(`ws://localhost:${port}`, {
     reconnect: true,
     maxReconnectAttempts: 10,
-    serviceName: 'test-service'
+    serviceName: 'test-service',
+    connectionCallback: () => {
+      client.createSubscription('query', {}, publish)
+      client.createSubscription('query', {}, publish)
+    }
   })
 
   function publish (data) {
@@ -176,9 +175,6 @@ test('subscription client multiple subscriptions is handled by one operation', t
     server.close()
     t.end()
   }
-
-  client.createSubscription('query', {}, publish)
-  client.createSubscription('query', {}, publish)
 })
 
 test('subscription client multiple subscriptions unsubscribe removes only one subscription', t => {
@@ -313,13 +309,14 @@ test('subscription client sending empty object payload on connection init', (t)
   const client = new SubscriptionClient(`ws://localhost:${port}`, {
     reconnect: true,
     maxReconnectAttempts,
-    serviceName: 'test-service'
-  })
-
-  client.createSubscription('query', {}, (data) => {
-    client.close()
-    server.close()
-    t.end()
+    serviceName: 'test-service',
+    connectionCallback: () => {
+      client.createSubscription('query', {}, (data) => {
+        client.close()
+        server.close()
+        t.end()
+      })
+    }
   })
 })
 
@@ -351,18 +348,114 @@ test('subscription client not throwing error on GQL_CONNECTION_KEEP_ALIVE type p
     maxReconnectAttempts: 10,
     serviceName: 'test-service',
     connectionCallback: () => {
+      client.createSubscription('query', {}, (data) => {
+        t.deepEqual(data, {
+          topic: 'test-service_1',
+          payload: null
+        })
+        clock.tick(200)
+        client.close()
+        t.end()
+      })
+
       clock.tick(200)
       clock.tick(200)
     }
   })
+})
 
-  client.createSubscription('query', {}, (data) => {
-    t.deepEqual(data, {
-      topic: 'test-service_1',
-      payload: null
+test('subscription client should throw on createSubscription if connection is not ready', (t) => {
+  const server = new WS.Server({ port: 0 })
+  const port = server.address().port
+
+  server.on('connection', function connection (ws) {
+    ws.on('message', function incoming (message) {
+      const data = JSON.parse(message)
+      if (data.type === 'connection_init') {
+        ws.send(JSON.stringify({ id: undefined, type: 'connection_error' }))
+      }
     })
-    clock.tick(200)
-    client.close()
-    t.end()
   })
+
+  const client = new SubscriptionClient(`ws://localhost:${port}`, {
+    reconnect: false,
+    maxReconnectAttempts: 0,
+    serviceName: 'test-service',
+    failedConnectionCallback: () => {
+      try {
+        client.createSubscription('query', {})
+      } catch (err) {
+        t.ok(err instanceof Error)
+      }
+      server.close()
+      client.close()
+      t.end()
+    }
+  })
+})
+
+test('subscription client should pass the error payload to failedConnectionCallback in case of a connection_error', (t) => {
+  const server = new WS.Server({ port: 0 })
+  const port = server.address().port
+  const errorPayload = { message: 'error' }
+
+  server.on('connection', function connection (ws) {
+    ws.on('message', function incoming (message) {
+      const data = JSON.parse(message)
+      if (data.type === 'connection_init') {
+        ws.send(JSON.stringify({ id: undefined, type: 'connection_error', payload: errorPayload }))
+      }
+    })
+  })
+
+  const client = new SubscriptionClient(`ws://localhost:${port}`, {
+    reconnect: false,
+    maxReconnectAttempts: 0,
+    serviceName: 'test-service',
+    failedConnectionCallback: (err) => {
+      t.deepEqual(err, errorPayload)
+
+      server.close()
+      client.close()
+      t.end()
+    }
+  })
+})
+
+test('subscription client does not send message if operation is already started', (t) => {
+  let sent = false
+  class MockSubscriptionClient extends SubscriptionClient {
+    sendMessage (operationId, type, payload) {
+      if (operationId && type === 'start') {
+        if (!sent) {
+          t.pass()
+          sent = true
+        } else {
+          t.fail('Should not send message if operation is already started')
+        }
+      }
+    }
+  }
+
+  const server = new WS.Server({ port: 0 })
+  const port = server.address().port
+
+  server.on('connection', function connection (ws) {
+    ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
+  })
+
+  const client = new MockSubscriptionClient(`ws://localhost:${port}`, {
+    reconnect: true,
+    maxReconnectAttempts: 10,
+    serviceName: 'test-service',
+    connectionCallback: async () => {
+      const operationId = client.createSubscription('query', {}, publish)
+      client.startOperation(operationId)
+      server.close()
+      client.close()
+      t.end()
+    }
+  })
+
+  function publish (data) { }
 })