onerror not called when gracefully closed client

carlhoerberg · carlhoerberg · commit 6a4b0f06e457 · 2022-08-03T20:02:25.000+02:00
diff --git a/src/amqp-base-client.ts b/src/amqp-base-client.ts
@@ -18,7 +18,7 @@ export abstract class AMQPBaseClient {
   channels: AMQPChannel[]
   protected connectPromise?: [(conn: AMQPBaseClient) => void, (err: Error) => void]
   protected closePromise?: [(value?: void) => void, (err: Error) => void]
-  closed = false
+  closed = true
   blocked?: string
   channelMax = 0
   frameMax: number
@@ -40,7 +40,6 @@ export abstract class AMQPBaseClient {
     if (name) this.name = name // connection name
     if (platform) this.platform = platform
     this.channels = [new AMQPChannel(this, 0)]
-    this.closed = true
     this.onerror = (error: AMQPError) => console.error("amqp-client connection closed", error.message)
     if (frameMax < 4096) throw new Error("frameMax must be 4096 or larger")
     this.frameMax = frameMax
diff --git a/src/amqp-socket-client.ts b/src/amqp-socket-client.ts
@@ -49,10 +49,6 @@ export class AMQPClient extends AMQPBaseClient {
     })
     return new Promise((resolve, reject) => {
       socket.on('error', (err) => reject(new AMQPError(err.message, this)))
-      socket.on('connect', () => {
-        socket.on('error', (err) => this.onerror(new AMQPError(err.message, this)))
-        socket.on('close', (hadError: boolean) => { if (!hadError) this.onerror(new AMQPError("Socket closed", this)) })
-      })
       this.connectPromise = [resolve, reject]
     })
   }
@@ -67,6 +63,12 @@ export class AMQPClient extends AMQPBaseClient {
     const sendStart = () => this.send(new Uint8Array([65, 77, 81, 80, 0, 0, 9, 1]))
     const conn = this.tls ? tls.connect(options, sendStart) : net.connect(options, sendStart)
     conn.on('data', this.onRead.bind(this))
+    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))
+      })
+    })
     return conn
   }
 
diff --git a/test/test.ts b/test/test.ts
@@ -592,3 +592,24 @@ test("can handle heartbeats", async t => {
   await wait
   t.is(conn.closed, false)
 })
+
+test("has an onerror callback", async t => {
+  const amqp = new AMQPClient("amqp://127.0.0.1")
+  const conn = await amqp.connect()
+  const ch = await conn.channel()
+  conn.onerror = (err) => {
+    t.regex(err.message, /invalid exchange type/)
+  }
+  await t.throwsAsync(ch.exchangeDeclare("none", "none"))
+})
+
+test("onerror is not called when conn is closed by client", async t => {
+  const amqp = new AMQPClient("amqp://127.0.0.1")
+  const conn = await amqp.connect()
+  conn.onerror = () => {
+    t.fail("onerror should not be called when gracefully closed")
+  }
+  await conn.close()
+  await new Promise((resolv) => setTimeout(resolv, 10))
+  t.pass()
+})
