Support negotiating channel max

baelter · baelter · commit f2ba1dffdc3a · 2023-08-23T11:09:23.000+02:00
diff --git a/src/amqp-base-client.ts b/src/amqp-base-client.ts
@@ -36,7 +36,7 @@ export abstract class AMQPBaseClient {
    * @param name - name of the connection, set in client properties
    * @param platform - used in client properties
    */
-  constructor(vhost: string, username: string, password: string, name?: string, platform?: string, frameMax = 4096, heartbeat = 0) {
+  constructor(vhost: string, username: string, password: string, name?: string, platform?: string, frameMax = 4096, heartbeat = 0, channelMax = 0) {
     this.vhost = vhost
     this.username = username
     this.password = ""
@@ -52,6 +52,8 @@ export abstract class AMQPBaseClient {
     this.frameMax = frameMax
     if (heartbeat < 0) throw new Error("heartbeat must be positive")
     this.heartbeat = heartbeat
+    if (channelMax && channelMax < 0) throw new Error("channelMax must be positive")
+    this.channelMax = channelMax
   }
 
   /**
@@ -222,7 +224,7 @@ export abstract class AMQPBaseClient {
                   const channelMax = view.getUint16(i); i += 2
                   const frameMax = view.getUint32(i); i += 4
                   const heartbeat = view.getUint16(i); i += 2
-                  this.channelMax = channelMax
+                  this.channelMax = this.channelMax === 0 ? channelMax : Math.min(this.channelMax, channelMax)
                   this.frameMax = this.frameMax === 0 ? frameMax : Math.min(this.frameMax, frameMax)
                   this.heartbeat = this.heartbeat === 0 ? 0 : Math.min(this.heartbeat, heartbeat)
 
diff --git a/src/amqp-socket-client.ts b/src/amqp-socket-client.ts
@@ -31,8 +31,9 @@ export class AMQPClient extends AMQPBaseClient {
     const name = u.searchParams.get("name") || ""
     const frameMax = parseInt(u.searchParams.get("frameMax") || "4096")
     const heartbeat = parseInt(u.searchParams.get("heartbeat") || "0")
+    const channelMax = parseInt(u.searchParams.get("channelMax") || "0")
     const platform = `${process.release.name} ${process.version} ${process.platform} ${process.arch}`
-    super(vhost, username, password, name, platform, frameMax, heartbeat)
+    super(vhost, username, password, name, platform, frameMax, heartbeat, channelMax)
     this.tls = u.protocol === "amqps:"
     this.tlsOptions = tlsOptions
     this.host = u.hostname || "localhost"
diff --git a/test/test.ts b/test/test.ts
@@ -3,10 +3,11 @@ import { AMQPClient } from '../src/amqp-socket-client.js';
 import { AMQPMessage } from '../src/amqp-message.js';
 import type { AMQPError } from "../src/amqp-error.js";
 
-function getNewClient(init?: {frameMax?: number, heartbeat?: number}): AMQPClient {
+function getNewClient(init?: { frameMax?: number, heartbeat?: number, channelMax?: number }): AMQPClient {
   const url = new URL("amqp://127.0.0.1")
   if (init?.frameMax != null) url.searchParams.append("frameMax", init.frameMax.toString())
   if (init?.heartbeat != null) url.searchParams.append("heartbeat", init.heartbeat.toString())
+  if (init?.channelMax != null) url.searchParams.append("channelMax", init.channelMax.toString())
 
   return new AMQPClient(url.toString())
 }
@@ -668,6 +669,13 @@ test("raises when channelMax is reached", async () => {
   await expect(ch1.basicQos(10)).resolves.toBeUndefined()
 }, 10_000)
 
+test("client can negotiate channelMax", async () => {
+  const amqp = getNewClient({ channelMax: 1 })
+  const conn = await amqp.connect()
+  await conn.channel()
+  await expect(conn.channel()).rejects.toThrow('Max number of channels reached');
+}, 10_000)
+
 test('can update-secret', async () => {
   const amqp = getNewClient()
   const conn = await amqp.connect()
