Support basicCancel from server

carlhoerberg · carlhoerberg · commit 3580798749d4 · 2022-01-31T23:08:42.000+01:00
Fixes #24
diff --git a/src/amqp-base-client.ts b/src/amqp-base-client.ts
@@ -408,6 +408,29 @@ export default abstract class AMQPBaseClient {
                   channel.resolvePromise(consumerTag)
                   break
                 }
+                case 30: { // cancel
+                  const [consumerTag, len] = view.getShortString(i); i += len
+                  const noWait = view.getUint8(i) === 1; i += 1
+
+                  const consumer = channel.consumers.get(consumerTag)
+                  if (consumer) {
+                    consumer.setClosed(new AMQPError("Consumer cancelled by the server", this))
+                    channel.consumers.delete(consumerTag)
+                  }
+                  if (!noWait) {
+                    const frame = new AMQPView(new ArrayBuffer(512))
+                    frame.setUint8(j, 1); j += 1 // type: method
+                    frame.setUint16(j, channel.id); j += 2 // channel
+                    frame.setUint32(j, 0); j += 4 // frameSize
+                    frame.setUint16(j, 60); j += 2 // class: basic
+                    frame.setUint16(j, 31); j += 2 // method: cancelOk
+                    j += frame.setShortString(j, consumerTag) // tag
+                    frame.setUint8(j, 206); j += 1 // frame end byte
+                    frame.setUint32(3, j - 8) // update frameSize
+                    this.send(new Uint8Array(frame.buffer, 0, j))
+                  }
+                  break
+                }
                 case 31: { // cancelOk
                   const [consumerTag, len] = view.getShortString(i); i += len
                   channel.resolvePromise(consumerTag)
diff --git a/test/test.ts b/test/test.ts
@@ -574,3 +574,13 @@ test("will raise if socket is closed on send", async t => {
   if (amqp.socket) amqp.socket.destroy()
   await t.throwsAsync(() => conn.channel())
 })
+
+test("can handle cancel from server", async t => {
+  const amqp = new AMQPClient("amqp://127.0.0.1")
+  const conn = await amqp.connect()
+  const ch = await conn.channel()
+  const q = await ch.queue("")
+  const consumer = await q.subscribe({}, () => "")
+  await q.delete()
+  await t.throwsAsync(() => consumer.wait(), { message: /Consumer cancelled by the server/ })
+})
