Support set connectionTimeout param (#410)

Author: shibd

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 @@ const Pulsar = require('../index');
         serviceUrl,
         tlsTrustCertsFilePath: `${__dirname}/certificate/server.crt`,
         operationTimeoutSeconds: 30,
+        connectionTimeoutMs: 20000,
         listenerName,
       });