Address code review & add more unit test

Author: shibd

index.d.ts

@@ -333,6 +333,7 @@ export type HashingScheme =
   'JavaStringHash';
 
 export type CompressionType =
+  'None' |
   'Zlib' |
   'LZ4' |
   'ZSTD' |

src/Message.cc

@@ -23,6 +23,7 @@
 #include <pulsar/Message.h>
 #include <pulsar/MessageBuilder.h>
 #include <pulsar/EncryptionContext.h>
+#include <map>
 
 struct _pulsar_message {
   pulsar::MessageBuilder builder;
@@ -40,6 +41,12 @@ static const std::string CFG_DELIVER_AT = "deliverAt";
 static const std::string CFG_DISABLE_REPLICATION = "disableReplication";
 static const std::string CFG_ORDERING_KEY = "orderingKey";
 
+static const std::map<pulsar::CompressionType, std::string> COMPRESSION_TYPE_MAP = {
+    {pulsar::CompressionNone, "None"},     {pulsar::CompressionLZ4, "LZ4"},
+    {pulsar::CompressionZLib, "Zlib"},     {pulsar::CompressionZSTD, "ZSTD"},
+    {pulsar::CompressionSNAPPY, "SNAPPY"},
+};
+
 Napi::FunctionReference Message::constructor;
 
 Napi::Object Message::Init(Napi::Env env, Napi::Object exports) {
@@ -183,12 +190,7 @@ Napi::Value Message::GetEncryptionContext(const Napi::CallbackInfo &info) {
   }
   const pulsar::EncryptionContext &encCtx = *encCtxPtr;
 
-  if (encCtx.keys().empty() && encCtx.param().empty() && encCtx.algorithm().empty()) {
-    return env.Null();
-  }
-
   Napi::Object obj = Napi::Object::New(env);
-
   Napi::Array keys = Napi::Array::New(env);
   int i = 0;
   for (const auto &keyInfo : encCtx.keys()) {
@@ -208,7 +210,9 @@ Napi::Value Message::GetEncryptionContext(const Napi::CallbackInfo &info) {
 
   obj.Set("param", Napi::Buffer<char>::Copy(env, encCtx.param().c_str(), encCtx.param().length()));
   obj.Set("algorithm", Napi::String::New(env, encCtx.algorithm()));
-  obj.Set("compressionType", Napi::Number::New(env, (int)encCtx.compressionType()));
+  const auto it = COMPRESSION_TYPE_MAP.find(encCtx.compressionType());
+  std::string compressionTypeStr = (it != COMPRESSION_TYPE_MAP.end()) ? it->second : "None";
+  obj.Set("compressionType", Napi::String::New(env, compressionTypeStr));
   obj.Set("uncompressedMessageSize", Napi::Number::New(env, encCtx.uncompressedMessageSize()));
   obj.Set("batchSize", Napi::Number::New(env, encCtx.batchSize()));
   obj.Set("isDecryptionFailed", Napi::Boolean::New(env, encCtx.isDecryptionFailed()));

src/ProducerConfig.cc

@@ -70,10 +70,8 @@ static const std::map<std::string, pulsar_hashing_scheme> HASHING_SCHEME = {
 };
 
 static std::map<std::string, pulsar_compression_type> COMPRESSION_TYPE = {
-    {"Zlib", pulsar_CompressionZLib},
-    {"LZ4", pulsar_CompressionLZ4},
-    {"ZSTD", pulsar_CompressionZSTD},
-    {"SNAPPY", pulsar_CompressionSNAPPY},
+    {"None", pulsar_CompressionNone}, {"Zlib", pulsar_CompressionZLib},     {"LZ4", pulsar_CompressionLZ4},
+    {"ZSTD", pulsar_CompressionZSTD}, {"SNAPPY", pulsar_CompressionSNAPPY},
 };
 
 static std::map<std::string, pulsar_producer_crypto_failure_action> PRODUCER_CRYPTO_FAILURE_ACTION = {

tests/encryption.test.js

@@ -102,12 +102,81 @@ class MyCryptoKeyReader extends Pulsar.CryptoKeyReader {
 
       const msg = await consumer.receive();
       expect(msg.getData().toString()).toBe(msgContent);
+      const encCtx = msg.getEncryptionContext();
+      expect(encCtx).not.toBeNull();
+      expect(encCtx.isDecryptionFailed).toBe(false);
+      expect(encCtx.keys).toBeDefined();
+      expect(encCtx.keys.length).toBeGreaterThan(0);
+      expect(encCtx.keys[0].value).toBeInstanceOf(Buffer);
+      expect(encCtx.param).toBeInstanceOf(Buffer);
+      expect(encCtx.algorithm).toBe('');
+      expect(encCtx.compressionType).toBe('None');
+      expect(encCtx.uncompressedMessageSize).toBe(0);
+      expect(encCtx.batchSize).toBe(1);
 
       await consumer.acknowledge(msg);
       await producer.close();
       await consumer.close();
     });
 
+    test('End-to-End Encryption with Batching and Compression', async () => {
+      const topic = `persistent://public/default/test-encryption-batch-compress-${Date.now()}`;
+
+      const cryptoKeyReader = new MyCryptoKeyReader(
+        { 'my-key': publicKeyPath },
+        { 'my-key': privateKeyPath },
+      );
+
+      const producer = await client.createProducer({
+        topic,
+        encryptionKeys: ['my-key'],
+        cryptoKeyReader,
+        cryptoFailureAction: 'FAIL',
+        batchingEnabled: true,
+        batchingMaxMessages: 10,
+        batchingMaxPublishDelayMs: 100,
+        compressionType: 'Zlib',
+      });
+
+      const consumer = await client.subscribe({
+        topic,
+        subscription: 'sub-encryption-batch-compress',
+        cryptoKeyReader,
+        cryptoFailureAction: 'CONSUME',
+        subscriptionInitialPosition: 'Earliest',
+      });
+
+      const numMessages = 10;
+      const sendPromises = [];
+      for (let i = 0; i < numMessages; i += 1) {
+        sendPromises.push(producer.send({
+          data: Buffer.from(`message-${i}`),
+        }));
+      }
+      await Promise.all(sendPromises);
+
+      for (let i = 0; i < numMessages; i += 1) {
+        const msg = await consumer.receive();
+        expect(msg.getData().toString()).toBe(`message-${i}`);
+        const encCtx = msg.getEncryptionContext();
+        expect(encCtx).not.toBeNull();
+        expect(encCtx.isDecryptionFailed).toBe(false);
+        expect(encCtx.keys).toBeDefined();
+        expect(encCtx.keys.length).toBeGreaterThan(0);
+        expect(encCtx.keys[0].value).toBeInstanceOf(Buffer);
+        expect(encCtx.param).toBeInstanceOf(Buffer);
+        expect(encCtx.algorithm).toBe('');
+        expect(encCtx.compressionType).toBe('Zlib');
+        expect(encCtx.uncompressedMessageSize).toBeGreaterThan(0);
+        expect(encCtx.batchSize).toBe(numMessages);
+
+        await consumer.acknowledge(msg);
+      }
+
+      await producer.close();
+      await consumer.close();
+    });
+
     test('Decryption Failure', async () => {
       const topic = `persistent://public/default/test-decryption-failure-${Date.now()}`;