feat(State Channels): Ping every 10 seconds to persist connection

es/channel/handlers.js

@@ -21,7 +21,8 @@ import {
   changeStatus,
   changeState,
   send,
-  emit
+  emit,
+  disconnect
 } from './internal'
 import { unpackTx } from '../tx/builder'
 
@@ -225,6 +226,7 @@ export function awaitingShutdownOnChainTx (channel, message, state) {
 export function awaitingLeave (channel, message, state) {
   if (message.method === 'channels.leave') {
     state.resolve({ channelId: message.params.channel_id, signedTx: message.params.data.state })
+    disconnect(channel)
     return { handler: channelClosed }
   }
   if (message.method === 'channels.error') {

es/channel/index.js

@@ -31,7 +31,8 @@ import {
   initialize,
   enqueueAction,
   send,
-  call
+  call,
+  disconnect as channelDisconnect
 } from './internal'
 import * as R from 'ramda'
 
@@ -52,6 +53,13 @@ function on (event, callback) {
   eventEmitters.get(this).on(event, callback)
 }
 
+/**
+ * Close the connection
+ */
+function disconnect () {
+  return channelDisconnect(this)
+}
+
 /**
  * Get current status
  *
@@ -558,7 +566,8 @@ const Channel = AsyncInit.compose({
     callContract,
     callContractStatic,
     getContractCall,
-    getContractState
+    getContractState,
+    disconnect
   }
 })
 

es/channel/internal.js

@@ -21,6 +21,9 @@ import * as R from 'ramda'
 import { pascalToSnake } from '../utils/string'
 import { awaitingConnection } from './handlers'
 
+const PING_TIMEOUT_MS = 10000
+const PONG_TIMEOUT_MS = 5000
+
 const options = new WeakMap()
 const status = new WeakMap()
 const state = new WeakMap()
@@ -33,6 +36,8 @@ const actionQueue = new WeakMap()
 const actionQueueLocked = new WeakMap()
 const sequence = new WeakMap()
 const rpcCallbacks = new WeakMap()
+const pingTimeoutId = new WeakMap()
+const pongTimeoutId = new WeakMap()
 
 function channelURL (url, params, endpoint = 'channel') {
   const paramString = R.join('&', R.values(R.mapObjIndexed((value, key) =>
@@ -147,6 +152,24 @@ function call (channel, method, params) {
   })
 }
 
+function ping (channel) {
+  const ws = websockets.get(channel)
+  if (ws.readyState === ws.OPEN) {
+    ws._connection.ping()
+    clearTimeout(pongTimeoutId.get(channel))
+    pongTimeoutId.set(channel, setTimeout(() => ws._connection.drop(), PONG_TIMEOUT_MS))
+  }
+}
+
+function disconnect (channel) {
+  const ws = websockets.get(channel)
+  if (ws.readyState === ws.OPEN) {
+    ws._connection.close()
+    clearTimeout(pongTimeoutId.get(channel))
+    clearTimeout(pingTimeoutId.get(channel))
+  }
+}
+
 function WebSocket (url, callbacks) {
   function fireOnce (target, key, always) {
     target[key] = (...args) => {
@@ -178,11 +201,18 @@ async function initialize (channel, channelOptions) {
   eventEmitters.set(channel, new EventEmitter())
   sequence.set(channel, 0)
   rpcCallbacks.set(channel, new Map())
-  websockets.set(channel, await WebSocket(wsUrl, {
+  const ws = await WebSocket(wsUrl, {
     onopen: () => changeStatus(channel, 'connected'),
     onclose: () => changeStatus(channel, 'disconnected'),
     onmessage: ({ data }) => onMessage(channel, data)
-  }))
+  })
+  ws._connection.on('pong', () => {
+    clearTimeout(pongTimeoutId.get(channel))
+    clearTimeout(pingTimeoutId.get(channel))
+    pingTimeoutId.set(channel, setTimeout(() => ping(channel), PING_TIMEOUT_MS))
+  })
+  pingTimeoutId.set(channel, setTimeout(() => ping(channel), PING_TIMEOUT_MS))
+  websockets.set(channel, ws)
 }
 
 export {
@@ -196,5 +226,6 @@ export {
   changeState,
   send,
   enqueueAction,
-  call
+  call,
+  disconnect
 }

test/integration/channel.js

@@ -83,6 +83,11 @@ describe('Channel', function () {
     await initiator.spend('6000000000000000', await responder.address())
   })
 
+  after(() => {
+    initiatorCh.disconnect()
+    responderCh.disconnect()
+  })
+
   beforeEach(() => {
     responderShouldRejectUpdate = false
   })
@@ -301,7 +306,8 @@ describe('Channel', function () {
     await Promise.all([waitForChannel(initiatorCh), waitForChannel(responderCh)])
     sinon.assert.notCalled(initiatorSign)
     sinon.assert.notCalled(responderSign)
-    await initiatorCh.leave()
+    initiatorCh.disconnect()
+    responderCh.disconnect()
   })
 
   it('can create a contract and accept', async () => {