19 - Adds frame max negotiation

src/connection.ts

@@ -30,7 +30,7 @@ import {
   PublishErrorListener,
   ResponseDecoder,
 } from "./response_decoder"
-import { removeFrom } from "./util"
+import { DEFAULT_FRAME_MAX, DEFAULT_UNLIMITED_FRAME_MAX, removeFrom } from "./util"
 import { WaitingResponse } from "./waiting_response"
 import { SubscribeResponse } from "./responses/subscribe_response"
 import { TuneResponse } from "./responses/tune_response"
@@ -62,7 +62,7 @@ export class Connection {
   private consumers = new Map<number, Consumer>()
   private compressions = new Map<CompressionType, Compression>()
 
-  constructor(private readonly logger: Logger) {
+  constructor(private readonly logger: Logger, private frameMax: number = DEFAULT_FRAME_MAX) {
     this.heartbeat = new Heartbeat(this, this.logger)
     this.compressions.set(CompressionType.None, NoneCompression.create())
     this.compressions.set(CompressionType.Gzip, GzipCompression.create())
@@ -89,7 +89,7 @@ export class Connection {
   }
 
   static connect(params: ConnectionParams, logger?: Logger): Promise<Connection> {
-    return new Connection(logger ?? new NullLogger()).start(params)
+    return new Connection(logger ?? new NullLogger(), params.frameMax).start(params)
   }
 
   public start(params: ConnectionParams): Promise<Connection> {
@@ -235,6 +235,10 @@ export class Connection {
     return this.consumers.size
   }
 
+  public get currentFrameMax() {
+    return this.frameMax
+  }
+
   public send(cmd: Request): Promise<void> {
     return new Promise((res, rej) => {
       const body = cmd.toBuffer()
@@ -326,7 +330,8 @@ export class Connection {
     const heartbeat = extractHeartbeatInterval(heartbeatInterval, tuneResponse)
 
     return new Promise((res, rej) => {
-      const request = new TuneRequest({ frameMax: tuneResponse.frameMax, heartbeat })
+      this.frameMax = this.calculateFrameMaxSizeFrom(tuneResponse.frameMax)
+      const request = new TuneRequest({ frameMax: this.frameMax, heartbeat })
       this.socket.write(request.toBuffer(), (err) => {
         this.logger.debug(`Write COMPLETED for cmd TUNE: ${inspect(tuneResponse)} - err: ${err}`)
         return err ? rej(err) : res({ heartbeat })
@@ -463,6 +468,12 @@ export class Connection {
       response.messages.map((x) => consumer.handle(x))
     })
   }
+
+  private calculateFrameMaxSizeFrom(tuneResponseFrameMax: number) {
+    if (this.frameMax === DEFAULT_UNLIMITED_FRAME_MAX) return tuneResponseFrameMax
+    if (tuneResponseFrameMax === DEFAULT_UNLIMITED_FRAME_MAX) return this.frameMax
+    return Math.min(this.frameMax, tuneResponseFrameMax)
+  }
 }
 
 export type ListenersParams = {
@@ -477,7 +488,7 @@ export interface ConnectionParams {
   username: string
   password: string
   vhost: string
-  frameMax?: number // not used
+  frameMax?: number
   heartbeat?: number
   listeners?: ListenersParams
 }

src/util.ts

@@ -31,3 +31,6 @@ export function range(count: number): number[] {
   }
   return ret
 }
+
+export const DEFAULT_FRAME_MAX = 1048576
+export const DEFAULT_UNLIMITED_FRAME_MAX = 0

test/e2e/connect_frame_size_negotiation.test.ts

@@ -0,0 +1,38 @@
+import { expect } from "chai"
+import { createConnection } from "../support/fake_data"
+import { Rabbit } from "../support/rabbit"
+import { eventually, username, password } from "../support/util"
+
+describe("connect frame size negotiation", () => {
+  const rabbit = new Rabbit(username, password)
+
+  it("using 65536 as frameMax", async () => {
+    const frameMax = 65536
+
+    const connection = await createConnection(username, password, undefined, frameMax)
+
+    await eventually(async () => {
+      expect(connection.currentFrameMax).lte(frameMax)
+      expect(await rabbit.getConnections()).lengthOf(1)
+    }, 5000)
+    try {
+      await connection.close()
+      await rabbit.closeAllConnections()
+    } catch (e) {}
+  }).timeout(10000)
+
+  it("using 1024 as frameMax", async () => {
+    const frameMax = 1024
+
+    const connection = await createConnection(username, password, undefined, frameMax)
+
+    await eventually(async () => {
+      expect(connection.currentFrameMax).lte(frameMax)
+      expect(await rabbit.getConnections()).lengthOf(1)
+    }, 5000)
+    try {
+      await connection.close()
+      await rabbit.closeAllConnections()
+    } catch (e) {}
+  }).timeout(10000)
+})

test/support/fake_data.ts

@@ -32,14 +32,14 @@ export async function createPublisher(streamName: string, connection: Connection
   return publisher
 }
 
-export function createConnection(username: string, password: string, listeners?: ListenersParams) {
+export function createConnection(username: string, password: string, listeners?: ListenersParams, frameMax?: number) {
   return connect({
     hostname: "localhost",
     port: 5552,
     username,
     password,
     vhost: "/",
-    frameMax: 0, // not used
+    frameMax: frameMax ?? 0,
     heartbeat: 0,
     listeners: listeners,
   })

test/unit/delete_stream.test.ts

@@ -15,8 +15,8 @@ describe("Delete command", () => {
       username,
       password,
       vhost: "/",
-      frameMax: 0, // not used
-      heartbeat: 0, // not user
+      frameMax: 0,
+      heartbeat: 0,
     })
   })