Implement proper connection loss handling for WebSocket connections

baelter · baelter · commit 96ede5d62ee4 · 2025-09-11T10:47:53.000+02:00
diff --git a/src/amqp-websocket-client.ts b/src/amqp-websocket-client.ts
@@ -59,11 +59,29 @@ export class AMQPWebSocketClient extends AMQPBaseClient {
       socket.addEventListener('close', reject)
       socket.addEventListener('error', reject)
       socket.addEventListener('open', () => {
-        socket.addEventListener('error', (ev: Event) => this.onerror(new AMQPError(ev.toString(), this)))
+        socket.addEventListener('error', (ev: Event) => {
+          if (!this.closed) {
+            const err = new AMQPError(ev.toString(), this)
+            this.closed = true
+            // Close all channels and their consumers when there's an error
+            this.channels.forEach((ch) => ch?.setClosed(err))
+            this.channels = [new AMQPChannel(this, 0)]
+            this.onerror(err)
+          }
+        })
         socket.addEventListener('close', (ev: CloseEvent) => {
           const clientClosed = this.closed
           this.closed = true
-          if (!ev.wasClean && !clientClosed) this.onerror(new AMQPError(`connection not cleanly closed (${ev.code})`, this))
+          if (!(ev.wasClean && clientClosed)) {
+            const err = new AMQPError(`connection not cleanly closed (${ev.code})`, this)
+            // Close all channels and their consumers when connection is lost
+            this.channels.forEach((ch) => ch?.setClosed(err))
+            this.channels = [new AMQPChannel(this, 0)]
+            this.onerror(err)
+          } else {
+            this.channels.forEach((ch) => ch?.setClosed())
+            this.channels = [new AMQPChannel(this, 0)]
+          }
         })
         socket.send(new Uint8Array([65, 77, 81, 80, 0, 0, 9, 1]))
       })
diff --git a/test-browser/websocket.ts b/test-browser/websocket.ts
@@ -226,6 +226,63 @@ test('closed socket closes client', async () => {
   expect(amqp.closed).toBe(true)
 })
 
+test('connection loss closes channels and consumers', async () => {
+  const amqp = getNewClient()
+  const conn = await amqp.connect()
+  const ch = await conn.channel()
+  const q = await ch.queue("")
+
+  const consumer = await q.subscribe({noAck: false}, () => {})
+
+  // Set up error handler to track when consumer is closed
+  const originalConsumerWait = consumer.wait()
+
+  const socket = amqp["socket"]
+  assert(socket, "Socket must be created")
+
+  // Simulate unclean connection loss by closing the socket
+  const closed = new Promise((resolve) => socket.addEventListener('close', resolve))
+  socket.close()
+  await closed
+
+  // Check that connection, channel, and consumer are all marked as closed
+  expect(amqp.closed).toBe(true)
+  expect(ch.closed).toBe(true)
+
+  // Consumer wait should reject with an error
+  await expect(originalConsumerWait).rejects.toThrow()
+
+  // Verify that operations on closed objects throw errors
+  await expect(ch.queue()).rejects.toThrow(/closed/)
+  await expect(q.publish("test")).rejects.toThrow()
+})
+
+test('connection loss triggers onerror callback', async () => {
+  const amqp = getNewClient()
+  const conn = await amqp.connect()
+
+  let errorReceived: AMQPError | null = null
+  conn.onerror = vi.fn((err: AMQPError) => {
+    errorReceived = err
+  })
+
+  const socket = amqp["socket"]
+  assert(socket, "Socket must be created")
+
+  // Simulate unclean connection loss
+  const closed = new Promise((resolve) => socket.addEventListener('close', resolve))
+  socket.close()
+  await closed
+
+  // Check that error callback was called
+  expect(conn.onerror).toHaveBeenCalled()
+  expect(errorReceived).toBeTruthy()
+  if (errorReceived) {
+    expect((errorReceived as AMQPError).message).toMatch(/connection not cleanly closed/)
+  }
+  expect(amqp.closed).toBe(true)
+})
+
 test('wait for publish confirms', async () => {
   const amqp = getNewClient()
   const conn = await amqp.connect()
