feat: implement graphql-ws ping/pong

lib/subscription-client.js

@@ -220,7 +220,11 @@ class SubscriptionClient {
         if (operation) {
           operation.handler(null)
           this.operations.delete(operationId)
-          this.sendMessage(operationId, this.protocolMessageTypes.GQL_ERROR, data.payload)
+          this.sendMessage(
+            operationId,
+            this.protocolMessageTypes.GQL_ERROR,
+            data.payload
+          )
         }
         break
       case this.protocolMessageTypes.GQL_COMPLETE:
@@ -238,11 +242,27 @@ class SubscriptionClient {
         }
         break
       case this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE:
+        if (this.socket) {
+          this.sendMessage(
+            operationId,
+            this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK
+          )
+        }
         break
-      /* istanbul ignore next */
       default:
+        // GQL_CONNECTION_KEEP_ALIVE_ACK is only defined in the graphql-ws protocol
+        /* istanbul ignore next */
+        if (
+          this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK &&
+          data.type === this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK
+        ) {
+          break
+        }
+
         /* istanbul ignore next */
-        throw new MER_ERR_GQL_SUBSCRIPTION_MESSAGE_INVALID(`Invalid message type "${data.type}"`)
+        throw new MER_ERR_GQL_SUBSCRIPTION_MESSAGE_INVALID(
+          `Invalid message type "${data.type}"`
+        )
     }
   }
 

lib/subscription-connection.js

@@ -74,20 +74,50 @@ module.exports = class SubscriptionConnection {
         break
       case this.protocolMessageTypes.GQL_START: {
         if (this.isReady) {
-          this.handleGQLStart(data).catch(e => {
-            this.sendMessage(this.protocolMessageTypes.GQL_ERROR, id, e.message)
+          this.handleGQLStart(data).catch((e) => {
+            this.sendMessage(
+              this.protocolMessageTypes.GQL_ERROR,
+              id,
+              e.message
+            )
           })
         } else {
-          this.sendMessage(this.protocolMessageTypes.GQL_CONNECTION_ERROR, undefined, { message: 'Connection has not been established yet.' })
+          this.sendMessage(
+            this.protocolMessageTypes.GQL_CONNECTION_ERROR,
+            undefined,
+            { message: 'Connection has not been established yet.' }
+          )
           return this.handleConnectionClose()
         }
         break
       }
       case this.protocolMessageTypes.GQL_STOP:
         await this.handleGQLStop(data)
         break
+      case this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE:
+        // GQL_CONNECTION_KEEP_ALIVE_ACK is only defined in the graphql-ws protocol
+        /* istanbul ignore next */
+        if (this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK) {
+          this.sendMessage(
+            this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK,
+            id
+          )
+        }
+        break
       default:
-        this.sendMessage(this.protocolMessageTypes.GQL_ERROR, id, 'Invalid payload type')
+        // GQL_CONNECTION_KEEP_ALIVE_ACK is only defined in the graphql-ws protocol
+        if (
+          this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK &&
+          type === this.protocolMessageTypes.GQL_CONNECTION_KEEP_ALIVE_ACK
+        ) {
+          break
+        }
+
+        this.sendMessage(
+          this.protocolMessageTypes.GQL_ERROR,
+          id,
+          'Invalid payload type'
+        )
     }
   }
 

lib/subscription-protocol.js

@@ -13,7 +13,8 @@ module.exports.getProtocolByName = function (name) {
         GQL_CONNECTION_INIT: 'connection_init', // Client -> Server
         GQL_CONNECTION_ACK: 'connection_ack', // Server -> Client
         GQL_CONNECTION_ERROR: 'connection_error', // Server -> Client
-        GQL_CONNECTION_KEEP_ALIVE: 'ka', // Server -> Client
+        GQL_CONNECTION_KEEP_ALIVE: 'ping', // Bidirectional
+        GQL_CONNECTION_KEEP_ALIVE_ACK: 'pong', // Bidirectional
         GQL_CONNECTION_TERMINATE: 'connection_terminate', // Client -> Server
         GQL_START: 'subscribe', // Client -> Server
         GQL_DATA: 'next', // Server -> Client

test/subscription-client.js

@@ -379,7 +379,14 @@ test('subscription client sends GQL_CONNECTION_KEEP_ALIVE when the keep alive op
         ws.send(JSON.stringify({ id: '1', type: 'connection_ack' }))
       } else if (data.type === 'start') {
         ws.send(JSON.stringify({ id: '2', type: 'complete' }))
-      } else if (data.type === 'ka') {
+      } else if (data.type === 'ping') {
+        // this is a client sent ping, we reply with our pong
+        ws.send(JSON.stringify({ id: '3', type: 'pong' }))
+
+        // send a ping to the client so that it replies with its own pong
+        // pings and pongs are bidirectional
+        ws.send(JSON.stringify({ id: '4', type: 'ping' }))
+      } else if (data.type === 'pong') {
         client.close()
         server.close()
         t.end()
@@ -414,7 +421,7 @@ test('subscription client not throwing error on GQL_CONNECTION_KEEP_ALIVE type p
     ws.send(JSON.stringify({ id: undefined, type: 'connection_ack' }))
 
     clock.setInterval(() => {
-      ws.send(JSON.stringify({ type: 'ka' }))
+      ws.send(JSON.stringify({ type: 'ping' }))
     }, 200)
   })
 

test/subscription-connection.js

@@ -341,6 +341,69 @@ test('subscription connection handles when GQL_START is called before GQL_INIT',
   }))
 })
 
+test('subscription connection replies to GQL_CONNECTION_KEEP_ALIVE message with GQL_CONNECTION_KEEP_ALIVE_ACK', async (t) => {
+  t.plan(1)
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      close () {},
+      send (message) {
+        t.equal(
+          JSON.stringify({
+            type: 'pong',
+            id: 1
+          }),
+          message
+        )
+      },
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {}
+  )
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'ping',
+      payload: {}
+    })
+  )
+})
+
+test('subscription connection does not error if client sends GQL_CONNECTION_KEEP_ALIVE_ACK', async (t) => {
+  t.plan(1)
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      close () {},
+      send (message) {
+        t.fail()
+      },
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {}
+  )
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'pong',
+      payload: {}
+    })
+  )
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'complete'
+    })
+  )
+
+  t.equal(sc.subscriptionContexts.size, 0)
+})
+
 test('subscription connection extends context with onConnect return value', async (t) => {
   t.plan(3)
 

test/subscription-protocol.js

@@ -20,7 +20,8 @@ test('getProtocolByName returns correct protocol message types', t => {
     GQL_CONNECTION_INIT: 'connection_init',
     GQL_CONNECTION_ACK: 'connection_ack',
     GQL_CONNECTION_ERROR: 'connection_error',
-    GQL_CONNECTION_KEEP_ALIVE: 'ka',
+    GQL_CONNECTION_KEEP_ALIVE: 'ping',
+    GQL_CONNECTION_KEEP_ALIVE_ACK: 'pong',
     GQL_CONNECTION_TERMINATE: 'connection_terminate',
     GQL_START: 'subscribe',
     GQL_DATA: 'next',