Return publish frame buffer to pool after send (#142)

carlhoerberg · antondalgren · web-flow · commit ea7bcca5fb37 · 2025-06-12T22:41:42.000+02:00
Fixes #137 --------- Co-authored-by: Anton Dalgren <antondalgren@users.noreply.github.com>
diff --git a/src/amqp-channel.ts b/src/amqp-channel.ts
@@ -363,15 +363,15 @@ export class AMQPChannel {
     }
     const sendFrames = this.connection.send(bufferView.subarray(0, j))
 
-    this.connection.bufferPool.push(buffer) // return buffer to buffer pool for later reuse
+    // return buffer to buffer pool for later reuse
     // if publish confirm is enabled, put a promise on a queue if the sends were ok
     // the promise on the queue will be fullfilled by the read loop when an ack/nack
     // comes from the server
     if (this.confirmId) {
       const wait4Confirm = new Promise<number>((resolve, reject) => this.unconfirmedPublishes.push([this.confirmId++, resolve, reject]))
-      return sendFrames.then(() => wait4Confirm)
+      return sendFrames.then(() => wait4Confirm).finally(() => this.connection.bufferPool.push(buffer))
     } else {
-      return sendFrames.then(() => 0)
+      return sendFrames.then(() => 0).finally(() => this.connection.bufferPool.push(buffer))
     }
   }
 
diff --git a/test/tls.ts b/test/tls.ts
@@ -1,5 +1,6 @@
 import { expect, test, beforeEach } from "vitest";
 import { AMQPClient } from '../src/amqp-socket-client.js';
+import { randomBytes } from 'crypto';
 
 beforeEach(() => {
   expect.hasAssertions()
@@ -11,3 +12,25 @@ test('can connect with TLS', () => {
     .then(conn => conn.channel())
     .then(ch => expect(ch.connection.channels.length).toBe(2)) // 2 because channel 0 is counted
 })
+
+test('can batch send message', async () => {
+  const messages = [
+    randomBytes(500),
+    randomBytes(500),
+    randomBytes(500),
+    randomBytes(500)
+  ]
+  const amqp = new AMQPClient("amqps://localhost?insecure=1")
+  const connection = await amqp.connect()
+  const channel = await connection.channel()
+  const queue = await channel.queue("bug137")
+  await channel.confirmSelect();
+  const sendMsgs = messages.map(message => queue.publish(JSON.stringify(message), {
+    contentType: 'application/json',
+    deliveryMode: 2,
+  }))
+
+  expect(connection['bufferPool'].length).toBe(0)
+  await Promise.all(sendMsgs)
+  expect(connection['bufferPool'].length).toBe(4)
+})

Original file line number	Diff line number	Diff line change
`@@ -363,15 +363,15 @@ export class AMQPChannel {`
`363`	`363`	`}`
`364`	`364`	`const sendFrames = this.connection.send(bufferView.subarray(0, j))`
`365`	`365`
`366`		`- this.connection.bufferPool.push(buffer) // return buffer to buffer pool for later reuse`
	`366`	`+ // return buffer to buffer pool for later reuse`
`367`	`367`	`// if publish confirm is enabled, put a promise on a queue if the sends were ok`
`368`	`368`	`// the promise on the queue will be fullfilled by the read loop when an ack/nack`
`369`	`369`	`// comes from the server`
`370`	`370`	`if (this.confirmId) {`
`371`	`371`	`const wait4Confirm = new Promise<number>((resolve, reject) => this.unconfirmedPublishes.push([this.confirmId++, resolve, reject]))`
`372`		`- return sendFrames.then(() => wait4Confirm)`
	`372`	`+ return sendFrames.then(() => wait4Confirm).finally(() => this.connection.bufferPool.push(buffer))`
`373`	`373`	`} else {`
`374`		`- return sendFrames.then(() => 0)`
	`374`	`+ return sendFrames.then(() => 0).finally(() => this.connection.bufferPool.push(buffer))`
`375`	`375`	`}`
`376`	`376`	`}`
`377`	`377`