Fix parallel queue binding issue by implementing proper RPC callback queue

Copilot · baelter · baelter · commit 19c60178db43 · 2025-09-11T08:42:13.000+02:00
Co-authored-by: baelter &lt;1399369+baelter@users.noreply.github.com&gt;
diff --git a/src/amqp-channel.ts b/src/amqp-channel.ts
@@ -14,6 +14,7 @@ export class AMQPChannel {
   readonly id: number
   readonly consumers = new Map<string, AMQPConsumer>()
   private rpcQueue: Promise<unknown> = Promise.resolve(true)
+  private readonly rpcCallbacks: [(value?: unknown) => void, (err?: Error) => void][] = []
   private readonly unconfirmedPublishes: [number, (confirmId: number) => void, (err?: Error) => void][] = []
   closed = false
   confirmId = 0
@@ -747,11 +748,20 @@ export class AMQPChannel {
   private sendRpc(frame: AMQPView, frameSize: number): Promise<any> {
     return new Promise((resolve, reject) => {
       this.rpcQueue = this.rpcQueue.then(() => {
-        this.resolveRPC = resolve
-        this.rejectRPC = reject
-        this.connection.send(new Uint8Array(frame.buffer, 0, frameSize))
-          .catch(reject)
-      })
+        // Add the callbacks to the queue before sending
+        this.rpcCallbacks.push([resolve, reject])
+        
+        return this.connection.send(new Uint8Array(frame.buffer, 0, frameSize))
+          .catch((err) => {
+            // Remove the callbacks from the queue if send fails
+            const callbacks = this.rpcCallbacks.pop()
+            if (callbacks) {
+              callbacks[1](err) // call reject
+            } else {
+              reject(err)
+            }
+          })
+      }).catch(reject)
     })
   }
 
@@ -770,7 +780,11 @@ export class AMQPChannel {
       this.closed = true
       this.consumers.forEach((consumer) => consumer.setClosed(err))
       this.consumers.clear()
-      this.rejectRPC(err)
+      
+      // Reject all pending RPC callbacks
+      this.rpcCallbacks.forEach(([, reject]) => reject(err))
+      this.rpcCallbacks.length = 0
+      
       // Reject and clear all unconfirmed publishes
       this.unconfirmedPublishes.forEach(([, , reject]) => reject(err))
       this.unconfirmedPublishes.length = 0
@@ -833,13 +847,25 @@ export class AMQPChannel {
    * Resolvs next RPC command
    * @ignore
    */
-  resolveRPC(value?: unknown) : unknown | void { return value }
+  resolveRPC(value?: unknown) : unknown | void {
+    const callbacks = this.rpcCallbacks.shift()
+    if (callbacks) {
+      callbacks[0](value) // call resolve
+    }
+    return value
+  }
 
   /**
    * Reject next RPC command
    * @ignore
    */
-  rejectRPC(err?: Error) : Error | void { return err }
+  rejectRPC(err?: Error) : Error | void {
+    const callbacks = this.rpcCallbacks.shift()
+    if (callbacks) {
+      callbacks[1](err) // call reject
+    }
+    return err
+  }
 
   /**
    * Deliver a message to a consumer
diff --git a/test/test.ts b/test/test.ts
@@ -729,3 +729,26 @@ test('should handle heartbeat timeout correctly', async () => {
   expect(error.message).toEqual('Heartbeat timeout')
   expect(conn.closed).toBe(true)
 })
+
+test('can bind queues in parallel', async () => {
+  const amqp = getNewClient()
+  const conn = await amqp.connect()
+  const ch = await conn.channel()
+  const q = await ch.queue('test-queue')
+  await ch.exchangeDeclare('test-exchange', 'fanout')
+
+  // This should not hang - parallel binds should work
+  await Promise.all([
+    q.bind('test-exchange', 'foo:*'),
+    q.bind('test-exchange', 'bar:*'),
+  ])
+
+  // Test with more binds to stress test the RPC queue
+  await Promise.all([
+    q.bind('test-exchange', 'baz:*'),
+    q.bind('test-exchange', 'qux:*'),
+    q.bind('test-exchange', 'quux:*'),
+  ])
+
+  await expect(Promise.resolve('success')).resolves.toBe('success')
+})
