Set client as closed when socket closes

ngbrown · baelter · commit db2d87421378 · 2023-08-23T11:11:10.000+02:00
diff --git a/src/amqp-socket-client.ts b/src/amqp-socket-client.ts
@@ -83,7 +83,9 @@ export class AMQPClient extends AMQPBaseClient {
     conn.on('connect', () => {
       conn.on('error', (err) => this.onerror(new AMQPError(err.message, this)))
       conn.on('close', (hadError: boolean) => {
-        if (!hadError && !this.closed) this.onerror(new AMQPError("Socket closed", this))
+        const clientClosed = this.closed
+        this.closed = true
+        if (!hadError && !clientClosed) this.onerror(new AMQPError("Socket closed", this))
       })
     })
     return conn
diff --git a/src/amqp-websocket-client.ts b/src/amqp-websocket-client.ts
@@ -57,7 +57,9 @@ export class AMQPWebSocketClient extends AMQPBaseClient {
       socket.addEventListener('open', () => {
         socket.addEventListener('error', (ev: Event) => this.onerror(new AMQPError(ev.toString(), this)))
         socket.addEventListener('close', (ev: CloseEvent) => {
-          if (!ev.wasClean && !this.closed) this.onerror(new AMQPError(`connection not cleanly closed (${ev.code})`, this))
+          const clientClosed = this.closed
+          this.closed = true
+          if (!ev.wasClean && !clientClosed) this.onerror(new AMQPError(`connection not cleanly closed (${ev.code})`, this))
         })
         socket.send(new Uint8Array([65, 77, 81, 80, 0, 0, 9, 1]))
       })
diff --git a/test-browser/websocket.ts b/test-browser/websocket.ts
@@ -213,6 +213,17 @@ test('connection error raises on publish', async () => {
   await expect(q.publish("foobar")).rejects.toThrow()
 })
 
+test('closed socket closes client', async () => {
+  const amqp = getNewClient()
+  await amqp.connect()
+  const socket = amqp["socket"]
+  assert(socket, "Socket must be created")
+  const closed = new Promise((resolve) => socket.addEventListener('close', resolve))
+  socket.close()
+  await closed
+  expect(amqp.closed).toBe(true)
+})
+
 test('wait for publish confirms', async () => {
   const amqp = getNewClient()
   const conn = await amqp.connect()
diff --git a/test/test.ts b/test/test.ts
@@ -218,6 +218,17 @@ test('connection error raises on publish', async () => {
   await expect(q.publish("foobar")).rejects.toThrow()
 })
 
+test('closed socket closes client', async () => {
+  const amqp = getNewClient()
+  await amqp.connect()
+  const socket = amqp["socket"]
+  assert(socket, "Socket must be created")
+  const closed = new Promise((resolve) => socket.on('close', resolve))
+  socket.destroy()
+  await closed
+  expect(amqp.closed).toBe(true)
+})
+
 test('wait for publish confirms', async () => {
   const amqp = getNewClient()
   const conn = await amqp.connect()
