feat(rtsp): add retry helper

Adds a helper that can be used to automatically retry the last RTSP
method if it failed. The number of retries are configurable.

lib/components/rtsp-session/index.test.ts

@@ -175,7 +175,7 @@ describe('session', () => {
           expect(retry.uri).toEqual(s.uri)
           done()
         })
-        s._retry()
+        s.retry()
       })
       s.send({ method: RTSP_METHOD.DESCRIBE })
     })
@@ -187,7 +187,7 @@ describe('session', () => {
           expect(retry.headers.CSeq).toEqual(req.headers.CSeq + 1)
           done()
         })
-        s._retry()
+        s.retry()
       })
       s.send({ method: RTSP_METHOD.DESCRIBE })
     })

lib/components/rtsp-session/index.ts

@@ -83,6 +83,16 @@ const defaultConfig = (
   return { uri }
 }
 
+export class RTSPResponseError extends Error {
+  public code: number
+
+  constructor(message: string, code: number) {
+    super(message)
+    this.name = 'RTSPResponseError'
+    this.code = code
+  }
+}
+
 /**
  * A component that sets up a command queue in order to interact with the RTSP
  * server. Allows control over the RTSP session by listening to incoming messages
@@ -106,12 +116,13 @@ export class RtspSession extends Tube {
   public startTime?: number
 
   public onSdp?: (sdp: Sdp) => void
-  public onError?: (err: Error) => void
+  public onError?: (err: RTSPResponseError) => void
   public onPlay?: (range?: string[]) => void
 
+  public retry?: () => void
+
   private _outgoingClosed: boolean
   private _sequence?: number
-  private _retry?: () => void
   private _callStack?: Command[]
   private _callHistory?: any[]
   private _state?: STATE
@@ -213,7 +224,7 @@ export class RtspSession extends Tube {
    */
   _reset() {
     this._sequence = 1
-    this._retry = () => console.error("No request sent, can't retry")
+    this.retry = () => console.error("No request sent, can't retry")
     this._callStack = []
     this._callHistory = []
     this._state = STATE.IDLE
@@ -293,7 +304,8 @@ export class RtspSession extends Tube {
     }
     if (status >= 400) {
       // TODO: Retry in certain cases?
-      this.onError && this.onError(new Error(msg.data.toString('ascii')))
+      this.onError &&
+        this.onError(new RTSPResponseError(msg.data.toString('ascii'), status))
     }
 
     if (method === RTSP_METHOD.PLAY) {
@@ -463,7 +475,7 @@ export class RtspSession extends Tube {
       throw new Error('missing method when send request')
     }
     this._waiting = true
-    this._retry = this.send.bind(this, cmd)
+    this.retry = this.send.bind(this, cmd)
 
     if (
       this._sequence === undefined ||

lib/utils/index.browser.ts

@@ -1,2 +1,3 @@
 export { Scheduler } from './scheduler'
 export * from './protocols'
+export * from './retry'

lib/utils/index.node.ts

@@ -1,3 +1,4 @@
 export { Scheduler } from './scheduler'
 export { Clock } from './clock'
 export * from './protocols'
+export * from './retry'

lib/utils/retry.ts

@@ -0,0 +1,33 @@
+import { RtspSession } from '../components/rtsp-session'
+
+/**
+ * Retry failed commands.
+ *
+ * This retries RTSP commands that fails up to a certain
+ * limit of times.
+ */
+export const addRTSPRetry = (
+  rtspSession: RtspSession,
+  { maxRetries, errors } = { maxRetries: 20, errors: [503] },
+) => {
+  let retries = 0
+
+  const oldOnError = rtspSession.onError
+
+  rtspSession.onError = (err) => {
+    oldOnError?.(err)
+
+    if (!errors.includes(err.code)) {
+      return
+    }
+
+    // Stop retrying after 20 tries (~20 seconds)
+    if ((retries += 1) > maxRetries) {
+      console.log('retry, too many', retries, maxRetries)
+      return
+    }
+
+    // Retry
+    setTimeout(() => rtspSession.retry?.(), retries * 100)
+  }
+}