Merge pull request #1 from github/should-retry

Delegate fatal socket close code handling

Author: dgraham

karma.config.cjs

@@ -1,4 +1,20 @@
-module.exports = function(config) {
+const ws = require('nodejs-websocket')
+
+ws.createServer(function (conn) {
+  conn.on('text', function (msg) {
+    if (msg.startsWith('echo:')) {
+      conn.sendText(msg.replace('echo:', ''))
+    } else if (msg.startsWith('close:')) {
+      const code = msg.replace('close:', '')
+      conn.close(code, 'reason')
+    }
+  })
+  conn.on('error', function (error) {
+    if (error.code !== 'ECONNRESET') throw error
+  })
+}).listen(7999)
+
+module.exports = function (config) {
   config.set({
     frameworks: ['mocha', 'chai'],
     files: [

package-lock.json

@@ -35,6 +35,7 @@
     "karma-mocha": "^2.0.1",
     "karma-mocha-reporter": "^2.2.5",
     "mocha": "^8.0.1",
+    "nodejs-websocket": "^1.7.2",
     "rollup": "^2.21.0",
     "typescript": "^3.9.5"
   },

package.json

@@ -1,4 +1,5 @@
 import type {Socket, SocketDelegate, StableSocket} from './stable-socket'
+import {isFatal} from './stable-socket'
 
 export class BufferedSocket implements Socket, SocketDelegate {
   private buf: string[] = []
@@ -55,4 +56,8 @@ export class BufferedSocket implements Socket, SocketDelegate {
   socketDidReceiveMessage(socket: Socket, message: string): void {
     this.delegate.socketDidReceiveMessage(socket, message)
   }
+
+  socketShouldRetry(socket: Socket, code: number): boolean {
+    return this.delegate.socketShouldRetry ? this.delegate.socketShouldRetry(socket, code) : !isFatal(code)
+  }
 }

src/buffered-socket.ts

@@ -18,6 +18,7 @@ export interface SocketDelegate {
   socketDidClose(socket: Socket, code?: number, reason?: string): void
   socketDidFinish(socket: Socket): void
   socketDidReceiveMessage(socket: Socket, message: string): void
+  socketShouldRetry?(socket: Socket, code: number): boolean
 }
 
 export class StableSocket implements Socket {
@@ -48,7 +49,10 @@ export class StableSocket implements Socket {
     this.socket.onclose = (event: CloseEvent) => {
       this.socket = null
       this.delegate.socketDidClose(this, event.code, event.reason)
-      if (isFatal(event.code)) {
+      const fatal = this.delegate.socketShouldRetry
+        ? !this.delegate.socketShouldRetry(this, event.code)
+        : isFatal(event.code)
+      if (fatal) {
         this.delegate.socketDidFinish(this)
       } else {
         setTimeout(() => this.open(), rand(100, 150))
@@ -88,7 +92,7 @@ function rand(min: number, max: number): number {
   return Math.random() * (max - min) + min
 }
 
-function isFatal(code: number): boolean {
+export function isFatal(code: number): boolean {
   return code === POLICY_VIOLATION || code === INTERNAL_ERROR
 }
 

src/stable-socket.ts

@@ -1,4 +1,70 @@
 import {retry, timeout, wait} from '../dist/async-tasks.js'
+import {StableSocket} from '../dist/index.js'
+
+class Delegate {
+  constructor(fatal) {
+    this.fatal = fatal
+    this.states = []
+  }
+  socketDidOpen() {
+    this.states.push('open')
+  }
+  socketDidClose() {
+    this.states.push('closed')
+  }
+  socketDidFinish() {
+    this.states.push('finished')
+  }
+  socketDidReceiveMessage(socket, message) {
+    this.states.push(`msg:${message}`)
+  }
+  socketShouldRetry(socket, code) {
+    return code !== this.fatal
+  }
+}
+
+describe('StableSocket', function () {
+  it('invokes lifecycle delegate methods', async function () {
+    const url = 'ws://localhost:7999'
+    const delegate = new Delegate(0)
+    const policy = {timeout: 100, attempts: 1, maxDelay: 100}
+    const socket = new StableSocket(url, delegate, policy)
+    await socket.open()
+    assert(socket.isOpen())
+    assert.deepEqual(['open'], delegate.states)
+    socket.send('echo:hello')
+    await wait(10)
+    socket.close()
+    assert.deepEqual(['open', 'msg:hello', 'closed', 'finished'], delegate.states)
+  })
+
+  it('retries on non-fatal close code', async function () {
+    const url = 'ws://localhost:7999'
+    const delegate = new Delegate(0)
+    const policy = {timeout: 100, attempts: 1, maxDelay: 100}
+    const socket = new StableSocket(url, delegate, policy)
+    await socket.open()
+    assert(socket.isOpen())
+    assert.deepEqual(['open'], delegate.states)
+    socket.send('close:1000')
+    await wait(200)
+    assert.deepEqual(['open', 'closed', 'open'], delegate.states)
+    socket.close()
+  })
+
+  it('does not retry on fatal close code', async function () {
+    const url = 'ws://localhost:7999'
+    const delegate = new Delegate(4000)
+    const policy = {timeout: 100, attempts: 1, maxDelay: 100}
+    const socket = new StableSocket(url, delegate, policy)
+    await socket.open()
+    assert(socket.isOpen())
+    assert.deepEqual(['open'], delegate.states)
+    socket.send('close:4000')
+    await wait(200)
+    assert.deepEqual(['open', 'closed', 'finished'], delegate.states)
+  })
+})
 
 describe('async-tasks', function () {
   describe('timeout', function () {