[fix][auth] Allow natural exit with token suppliers

src/Authentication.cc

@@ -26,8 +26,6 @@ static const std::string PARAM_TOKEN = "token";
 static const std::string PARAM_USERNAME = "username";
 static const std::string PARAM_PASSWORD = "password";
 
-void FinalizeTokenSupplierCallback(Napi::Env env, TokenSupplierCallback *cb, void *) { delete cb; }
-
 struct TokenSupplierProxyData {
   std::function<void(void)> callback;
   std::string token;
@@ -134,14 +132,16 @@ Authentication::Authentication(const Napi::CallbackInfo &info)
         }
 
         if (obj.Get(PARAM_TOKEN).IsFunction()) {
-          this->tokenSupplier = new TokenSupplierCallback();
+          this->tokenSupplier = std::make_shared<TokenSupplierCallback>();
+          std::shared_ptr<TokenSupplierCallback> tokenSupplier = this->tokenSupplier;
           Napi::ThreadSafeFunction callback = Napi::ThreadSafeFunction::New(
               obj.Env(), obj.Get(PARAM_TOKEN).As<Napi::Function>(), "Token Supplier Callback", 1, 1,
-              (void *)NULL, FinalizeTokenSupplierCallback, this->tokenSupplier);
+              [tokenSupplier](Napi::Env) { tokenSupplier->finalized = true; });
           this->tokenSupplier->callback = std::move(callback);
+          this->tokenSupplier->callback.Unref(env);
 
           this->cAuthentication =
-              pulsar_authentication_token_create_with_supplier(&TokenSupplier, this->tokenSupplier);
+              pulsar_authentication_token_create_with_supplier(&TokenSupplier, this->tokenSupplier.get());
           return;
         }
       }
@@ -178,7 +178,7 @@ Authentication::Authentication(const Napi::CallbackInfo &info)
 }
 
 Authentication::~Authentication() {
-  if (this->tokenSupplier != nullptr) {
+  if (this->tokenSupplier != nullptr && !this->tokenSupplier->finalized) {
     this->tokenSupplier->callback.Release();
   }
   if (this->cAuthentication != nullptr) {

src/Authentication.h

@@ -20,6 +20,7 @@
 #ifndef AUTH_H
 #define AUTH_H
 
+#include <memory>
 #include <napi.h>
 #include <pulsar/c/authentication.h>
 #include "TokenSupplier.h"
@@ -35,7 +36,7 @@ class Authentication : public Napi::ObjectWrap<Authentication> {
   static Napi::FunctionReference constructor;
   pulsar_authentication_t *cAuthentication;
 
-  TokenSupplierCallback *tokenSupplier;
+  std::shared_ptr<TokenSupplierCallback> tokenSupplier;
 };
 
 #endif

src/Client.cc

@@ -62,7 +62,8 @@ 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;
+std::shared_ptr<LogCallback> Client::logCallback;
+std::mutex Client::logMutex;
 
 struct _pulsar_client_configuration {
   pulsar::ClientConfiguration conf;
@@ -259,27 +260,34 @@ static std::unique_ptr<pulsar::ServiceInfoProvider> BuildServiceInfoProvider(
 void Client::SetLogHandler(const Napi::CallbackInfo &info) {
   Napi::Env env = info.Env();
   Napi::HandleScope scope(env);
-  Napi::Value jsFunction = info[0];
+  ConfigureLogHandler(info[0]);
+}
 
-  if (jsFunction.IsNull()) {
-    if (Client::logCallback != nullptr) {
-      Client::logCallback->callback.Release();
-      delete Client::logCallback;
-      Client::logCallback = nullptr;
-    }
-  } else if (jsFunction.IsFunction()) {
-    Napi::ThreadSafeFunction logFunction =
-        Napi::ThreadSafeFunction::New(env, jsFunction.As<Napi::Function>(), "Pulsar Logging", 0, 1);
-    logFunction.Unref(env);
-    if (Client::logCallback != nullptr) {
-      Client::logCallback->callback.Release();
-    } else {
-      Client::logCallback = new LogCallback();
-    }
-    Client::logCallback->callback = std::move(logFunction);
-  } else {
+void Client::ConfigureLogHandler(Napi::Value jsFunction) {
+  Napi::Env env = jsFunction.Env();
+  if (!jsFunction.IsNull() && !jsFunction.IsFunction()) {
     Napi::Error::New(env, "Client log handler must be a function or null").ThrowAsJavaScriptException();
+    return;
+  }
+
+  std::shared_ptr<LogCallback> next;
+  if (jsFunction.IsFunction()) {
+    next = std::make_shared<LogCallback>();
+    next->callback = Napi::ThreadSafeFunction::New(env, jsFunction.As<Napi::Function>(), "Pulsar Logging", 0,
+                                                   1, [next](Napi::Env) {
+                                                     std::lock_guard<std::mutex> lock(logMutex);
+                                                     if (logCallback == next) {
+                                                       logCallback.reset();
+                                                     }
+                                                   });
+    next->callback.Unref(env);
   }
+
+  std::lock_guard<std::mutex> lock(logMutex);
+  if (logCallback != nullptr) {
+    logCallback->callback.Release();
+  }
+  logCallback = std::move(next);
 }
 
 Napi::FunctionReference Client::constructor;
@@ -352,16 +360,7 @@ Client::Client(const Napi::CallbackInfo &info) : Napi::ObjectWrap<Client>(info)
 
   // log config option should be deprecated in favour of static setLogHandler method
   if (clientConfig.Has(CFG_LOG) && clientConfig.Get(CFG_LOG).IsFunction()) {
-    Napi::ThreadSafeFunction logFunction = Napi::ThreadSafeFunction::New(
-        env, clientConfig.Get(CFG_LOG).As<Napi::Function>(), "Pulsar Logging", 0, 1);
-    logFunction.Unref(env);
-
-    if (Client::logCallback != nullptr) {
-      Client::logCallback->callback.Release();
-    } else {
-      Client::logCallback = new LogCallback();
-    }
-    Client::logCallback->callback = std::move(logFunction);
+    ConfigureLogHandler(clientConfig.Get(CFG_LOG));
   }
 
   if (clientConfig.Has(CFG_AUTH) && clientConfig.Get(CFG_AUTH).IsObject()) {
@@ -535,7 +534,10 @@ void LogMessageProxy(Napi::Env env, Napi::Function jsCallback, struct LogMessage
 
 void Client::LogMessage(pulsar_logger_level_t level, const char *file, int line, const char *message,
                         void *ctx) {
-  LogCallback *logCallback = Client::logCallback;
+  // The logging TSFN uses maxQueueSize = 0, so BlockingCall does not wait
+  // for JS to drain the queue while logMutex is held.
+  std::lock_guard<std::mutex> lock(logMutex);
+  auto logCallback = Client::logCallback;
 
   if (logCallback == nullptr) {
     return;
@@ -547,8 +549,13 @@ void Client::LogMessage(pulsar_logger_level_t level, const char *file, int line,
 
   struct LogMessage *logMessage = new struct LogMessage(level, std::string(file), line, std::string(message));
 
-  logCallback->callback.BlockingCall(logMessage, LogMessageProxy);
-  logCallback->callback.Release();
+  napi_status status = logCallback->callback.BlockingCall(logMessage, LogMessageProxy);
+  if (status != napi_ok) {
+    delete logMessage;
+  }
+  if (status != napi_closing) {
+    logCallback->callback.Release();
+  }
 }
 
 Napi::Value Client::Close(const Napi::CallbackInfo &info) {

src/Client.h

@@ -20,6 +20,8 @@
 #ifndef CLIENT_H
 #define CLIENT_H
 
+#include <memory>
+#include <mutex>
 #include <napi.h>
 #include <pulsar/c/client.h>
 #include <vector>
@@ -50,13 +52,15 @@ class Client : public Napi::ObjectWrap<Client> {
   ~Client();
 
  private:
-  static LogCallback *logCallback;
+  static std::shared_ptr<LogCallback> logCallback;
+  static std::mutex logMutex;
   static Napi::FunctionReference constructor;
   std::shared_ptr<pulsar_client_t> cClient;
   std::shared_ptr<pulsar_client_configuration_t> cClientConfig;
   pulsar_logger_level_t logLevel = pulsar_logger_level_t::pulsar_INFO;
   std::vector<Napi::ObjectReference> authRefs_;
 
+  static void ConfigureLogHandler(Napi::Value jsFunction);
   Napi::Value CreateProducer(const Napi::CallbackInfo &info);
   Napi::Value Subscribe(const Napi::CallbackInfo &info);
   Napi::Value CreateReader(const Napi::CallbackInfo &info);

src/TokenSupplier.h

@@ -24,6 +24,7 @@
 
 struct TokenSupplierCallback {
   Napi::ThreadSafeFunction callback;
+  bool finalized = false;
 
   TokenSupplierCallback() {}
 };

tests/process_exit.test.js

@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const path = require('path');
+const { spawn } = require('child_process');
+
+const scenariosDir = path.join(__dirname, 'process_exit');
+const runner = path.join(scenariosDir, 'runner.js');
+
+const expectNaturalExit = (script, args) => new Promise((resolve, reject) => {
+  const child = spawn(process.execPath, [runner, script, ...args], {
+    stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
+  });
+  let stdout = '';
+  let stderr = '';
+  let spawnError;
+  let protocolFailure;
+  let timedOut;
+  let completed = false;
+  let disconnectedAfterCompletion = false;
+  let naturalExitTimer;
+
+  const watchdogTimer = setTimeout(() => {
+    timedOut = completed ? 'IPC disconnect' : 'scenario completion';
+    child.kill('SIGKILL');
+  }, 60000);
+
+  child.stdout.on('data', (data) => {
+    stdout += data.toString();
+  });
+  child.stderr.on('data', (data) => {
+    stderr += data.toString();
+  });
+  child.on('message', (message) => {
+    if (message && message.type === 'scenario-complete') {
+      completed = true;
+    }
+  });
+  child.on('disconnect', () => {
+    if (!completed) {
+      protocolFailure = 'IPC disconnected before scenario completion';
+      child.kill('SIGKILL');
+      return;
+    }
+
+    disconnectedAfterCompletion = true;
+    clearTimeout(watchdogTimer);
+    naturalExitTimer = setTimeout(() => {
+      timedOut = 'natural exit after scenario completion and IPC disconnect';
+      child.kill('SIGKILL');
+    }, 10000);
+  });
+  child.on('error', (error) => {
+    spawnError = error;
+    clearTimeout(watchdogTimer);
+    clearTimeout(naturalExitTimer);
+  });
+  child.on('close', (code, signal) => {
+    clearTimeout(watchdogTimer);
+    clearTimeout(naturalExitTimer);
+    if (timedOut || spawnError || protocolFailure || !completed
+      || !disconnectedAfterCompletion || code !== 0 || signal !== null) {
+      reject(new Error(`Child did not exit naturally: ${JSON.stringify({
+        timedOut,
+        spawnError: spawnError && spawnError.message,
+        protocolFailure,
+        completed,
+        disconnectedAfterCompletion,
+        code,
+        signal,
+        stdout,
+        stderr,
+      })}`));
+      return;
+    }
+    resolve();
+  });
+});
+
+describe('AuthenticationToken supplier process exit', () => {
+  const scenario = path.join(scenariosDir, 'token_supplier.js');
+
+  test.each(['sync', 'async'])('%s supplier exits after closing a client', async (supplier) => {
+    await expectNaturalExit(scenario, [supplier]);
+  }, 75000);
+});

tests/process_exit/runner.js

@@ -0,0 +1,53 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const disconnect = () => {
+  if (process.connected) {
+    process.disconnect();
+  }
+};
+
+const fail = (error) => {
+  console.error(error);
+  process.exitCode = 1;
+  disconnect();
+};
+
+const runScenario = async () => {
+  if (typeof process.send !== 'function') {
+    throw new Error('Scenario runner requires an IPC channel');
+  }
+
+  // eslint-disable-next-line global-require, import/no-dynamic-require
+  const scenario = require(process.argv[2]);
+  if (typeof scenario !== 'function') {
+    throw new TypeError('Scenario module must export a function');
+  }
+  await scenario(...process.argv.slice(3));
+};
+
+runScenario().then(() => {
+  process.send({ type: 'scenario-complete' }, (error) => {
+    if (error) {
+      fail(error);
+      return;
+    }
+    disconnect();
+  });
+}).catch(fail);

tests/process_exit/token_supplier.js

@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+const Pulsar = require('../../index');
+
+const suppliers = {
+  sync: () => 'dummy-token',
+  async: async () => 'dummy-token',
+};
+
+module.exports = async (mode) => {
+  const supplier = suppliers[mode];
+  if (!supplier) {
+    throw new Error(`Unknown supplier: ${mode}`);
+  }
+
+  global.auth = new Pulsar.AuthenticationToken({ token: supplier });
+  const client = new Pulsar.Client({
+    serviceUrl: 'pulsar://localhost:6650',
+    authentication: global.auth,
+  });
+  await client.close();
+};