54 ability to publish a batch of messages with gzip compression

src/compression.ts

@@ -1,6 +1,8 @@
+import { gunzipSync, gzipSync } from "node:zlib"
+
 export enum CompressionType {
   None = 0,
-  GZip = 1,
+  Gzip = 1,
   // Not implemented by default.
   // It is possible to add custom codec with StreamCompressionCodecs
   Snappy = 2,
@@ -31,3 +33,21 @@ export class NoneCompression implements Compression {
     return data
   }
 }
+
+export class GzipCompression implements Compression {
+  static create(): GzipCompression {
+    return new GzipCompression()
+  }
+
+  getType(): CompressionType {
+    return CompressionType.Gzip
+  }
+
+  compress(data: Buffer): Buffer {
+    return gzipSync(data)
+  }
+
+  decompress(data: Buffer): Buffer {
+    return gunzipSync(data)
+  }
+}

src/connection.ts

@@ -48,7 +48,7 @@ import { QueryOffsetResponse } from "./responses/query_offset_response"
 import { QueryOffsetRequest } from "./requests/query_offset_request"
 import { StoreOffsetRequest } from "./requests/store_offset_request"
 import { Logger, NullLogger } from "./logger"
-import { Compression, CompressionType, NoneCompression } from "./compression"
+import { Compression, CompressionType, GzipCompression, NoneCompression } from "./compression"
 
 export class Connection {
   private readonly socket = new Socket()
@@ -66,6 +66,7 @@ export class Connection {
     this.heartbeat = new Heartbeat(this, this.logger)
     this.decoder = new ResponseDecoder((...args) => this.responseReceived(...args), this.logger)
     this.compressions.set(CompressionType.None, NoneCompression.create())
+    this.compressions.set(CompressionType.Gzip, GzipCompression.create())
   }
 
   getCompression(compressionType: CompressionType) {

test/e2e/sub_entry_publish.test.ts

@@ -4,6 +4,7 @@ import { Producer } from "../../src/producer"
 import { createConnection, createPublisher, createStreamName } from "../support/fake_data"
 import { Rabbit } from "../support/rabbit"
 import { eventually, username, password } from "../support/util"
+import { CompressionType } from "../../src/compression"
 
 describe("publish a batch of messages", () => {
   const rabbit = new Rabbit(username, password)
@@ -42,4 +43,20 @@ describe("publish a batch of messages", () => {
       expect(info.messages).eql(messages.length)
     }, 10000)
   }).timeout(10000)
+
+  it("publish a batch of messages with compression", async () => {
+    const messages = [
+      { content: Buffer.from("Ciao") },
+      { content: Buffer.from("Ciao1") },
+      { content: Buffer.from("Ciao2") },
+      { content: Buffer.from("Ciao3") },
+    ]
+
+    await publisher.sendSubEntries(messages, CompressionType.Gzip)
+
+    await eventually(async () => {
+      const info = await rabbit.getQueueInfo(streamName)
+      expect(info.messages).eql(messages.length)
+    }, 10000)
+  }).timeout(10000)
 })