Support set connectionTimeout param

index.d.ts

@@ -33,6 +33,7 @@ export interface ClientConfig {
   listenerName?: string;
   log?: (level: LogLevel, file: string, line: number, message: string) => void;
   logLevel?: LogLevel;
+  connectionTimeoutMs?: number;
 }
 
 export class Client {

src/Client.cc

@@ -26,6 +26,7 @@
 #include <pulsar/c/client.h>
 #include <pulsar/c/client_configuration.h>
 #include <pulsar/c/result.h>
+#include "pulsar/ClientConfiguration.h"
 
 static const std::string CFG_SERVICE_URL = "serviceUrl";
 static const std::string CFG_AUTH = "authentication";
@@ -42,9 +43,14 @@ static const std::string CFG_STATS_INTERVAL = "statsIntervalInSeconds";
 static const std::string CFG_LOG = "log";
 static const std::string CFG_LOG_LEVEL = "logLevel";
 static const std::string CFG_LISTENER_NAME = "listenerName";
+static const std::string CFG_CONNECTION_TIMEOUT = "connectionTimeoutMs";
 
 LogCallback *Client::logCallback = nullptr;
 
+struct _pulsar_client_configuration {
+  pulsar::ClientConfiguration conf;
+};
+
 void Client::SetLogHandler(const Napi::CallbackInfo &info) {
   Napi::Env env = info.Env();
   Napi::HandleScope scope(env);
@@ -157,6 +163,13 @@ Client::Client(const Napi::CallbackInfo &info) : Napi::ObjectWrap<Client>(info)
     }
   }
 
+  if (clientConfig.Has(CFG_CONNECTION_TIMEOUT) && clientConfig.Get(CFG_CONNECTION_TIMEOUT).IsNumber()) {
+    int32_t connectionTimeoutMs = clientConfig.Get(CFG_CONNECTION_TIMEOUT).ToNumber().Int32Value();
+    if (connectionTimeoutMs > 0) {
+      cClientConfig.get()->conf.setConnectionTimeout(connectionTimeoutMs);
+    }
+  }
+
   if (clientConfig.Has(CFG_LISTENER_THREADS) && clientConfig.Get(CFG_LISTENER_THREADS).IsNumber()) {
     int32_t messageListenerThreads = clientConfig.Get(CFG_LISTENER_THREADS).ToNumber().Int32Value();
     if (messageListenerThreads > 0) {

tests/end_to_end.test.js

@@ -32,6 +32,7 @@
         serviceUrl,
         tlsTrustCertsFilePath: `${__dirname}/certificate/server.crt`,
         operationTimeoutSeconds: 30,
+        connectionTimeoutMs: 20000,
         listenerName,
       });
 
@@ -310,7 +311,7 @@
       });
 
       let consumer2Recv = 0;
       while (true) {
         try {
           const msg = await consumer2.receive(3000);
           await new Promise((resolve) => setTimeout(resolve, 10));
@@ -355,7 +356,7 @@
         topic,
         startMessageId: Pulsar.MessageId.earliest(),
         receiverQueueSize: 10,
         listener: async (message, reader) => {
           await new Promise((resolve) => setTimeout(resolve, 10));
           reader1Recv += 1;
         },
@@ -387,7 +388,7 @@
       await client.close();
     });
 
     test('Message Listener error handling', async () => {
       const client = new Pulsar.Client({
         serviceUrl: 'pulsar://localhost:6650',
       });
@@ -423,7 +424,7 @@
         subscription: 'sync',
         subscriptionType: 'Shared',
         subscriptionInitialPosition: 'Earliest',
         listener: (message, messageConsumer) => {
           throw new Error('consumer1 callback expected error');
         },
       });
@@ -433,7 +434,7 @@
         subscription: 'async',
         subscriptionType: 'Shared',
         subscriptionInitialPosition: 'Earliest',
         listener: async (message, messageConsumer) => {
           throw new Error('consumer2 callback expected error');
         },
       });