Make time duration since last server data available

Author: ngbrown

src/amqp-base-client.ts

@@ -25,6 +25,7 @@ export abstract class AMQPBaseClient {
   channelMax = 0
   frameMax: number
   heartbeat: number
+  lastDataReceived: number | undefined = undefined
   onerror: (error: AMQPError) => void
   logger: Logger | null | undefined = console
   /** Used for string -> arraybuffer when publishing */
@@ -85,6 +86,7 @@ export abstract class AMQPBaseClient {
   close(reason = "", code = 200): Promise<void> {
     if (this.closed) return this.rejectClosed()
     this.closed = true
+    this.lastDataReceived = undefined
     let j = 0
     const frame = new AMQPView(new ArrayBuffer(512))
     frame.setUint8(j, 1); j += 1 // type: method
@@ -606,5 +608,18 @@ export abstract class AMQPBaseClient {
       }
       i += 1 // frame end
     }
+    this.lastDataReceived = performance.now()
+  }
+
+  /**
+   * Get the time since since connection or last data received
+   * @returns milliseconds; Infinity if disconnected
+   */
+  durationSinceLastData(): number {
+    if (typeof this.lastDataReceived === 'number') {
+      return performance.now() - this.lastDataReceived;
+    } else {
+      return Infinity
+    }
   }
 }

src/amqp-socket-client.ts

@@ -58,6 +58,7 @@ export class AMQPClient extends AMQPBaseClient {
     socket.setTimeout((this.heartbeat || 60) * 1000)
     // enable TCP keepalive if AMQP heartbeats are disabled
     if (this.heartbeat === 0) socket.setKeepAlive(true, 60)
+    this.lastDataReceived = performance.now()
     return new Promise((resolve, reject) => {
       socket.on('timeout', () => reject(new AMQPError("timeout", this)))
       socket.on('error', (err) => reject(new AMQPError(err.message, this)))

src/amqp-websocket-client.ts

@@ -50,6 +50,7 @@ export class AMQPWebSocketClient extends AMQPBaseClient {
     this.socket = socket
     socket.binaryType = "arraybuffer"
     socket.onmessage = this.handleMessage.bind(this)
+    this.lastDataReceived = performance.now()
     return new Promise((resolve, reject) => {
       this.connectPromise = [resolve, reject]
       socket.addEventListener('close', reject)