Properly handle heartbeat timeouts (#95)

carlhoerberg · carlhoerberg · commit 1a407f4f02b1 · 2025-05-28T11:40:02.000+02:00
On heartbeat timeout only the next RPC action was rejected, now `onerror` is called and the socket is closed. Fixes #95
diff --git a/src/amqp-socket-client.ts b/src/amqp-socket-client.ts
@@ -78,6 +78,10 @@ export class AMQPClient extends AMQPBaseClient {
       socket.on('error', (err) => reject(new AMQPError(err.message, this)))
       const onConnect = (conn: AMQPBaseClient) => {
         socket.setTimeout(this.heartbeat * 1000) // reset timeout if heartbeats are disabled
+        socket.on('timeout', () => {
+          this.onerror(new AMQPError(`Heartbeat timeout`, this))
+          this.closeSocket()
+        })
         resolve(conn)
       }
       this.connectPromise = [onConnect, reject]
diff --git a/test/test.ts b/test/test.ts
@@ -697,3 +697,35 @@ test('should fail to connect to HTTP', async () => {
   const amqp = new AMQPClient("amqp://127.0.0.1:15672?heartbeat=1")
   await expect(amqp.connect()).rejects.toThrow()
 })
+
+test('should handle heartbeat timeout correctly', async () => {
+  const amqp = getNewClient({ heartbeat: 1 })
+  const conn = await amqp.connect()
+  
+  // Mock the socket timeout to simulate missed heartbeats
+  const socket = amqp["socket"]
+  assert(socket, "Socket must be created")
+  
+  // Set up error callback to capture timeout error
+  const errorPromise = new Promise<AMQPError>((resolve) => {
+    conn.onerror = (err: AMQPError) => {
+      resolve(err)
+    }
+  })
+  
+  // Set up close promise to detect when socket is closed
+  const closePromise = new Promise((resolve) => {
+    socket.on('close', resolve)
+  })
+  
+  // Trigger timeout event to simulate heartbeat timeout
+  socket.emit('timeout')
+  
+  // Wait for error callback and socket close
+  const error = await errorPromise
+  await closePromise
+  
+  // Verify error message and that connection is closed
+  expect(error.message).toEqual('Heartbeat timeout')
+  expect(conn.closed).toBe(true)
+})
