diff --git a/src/Authentication.cc b/src/Authentication.cc index 3b04554..623d59c 100644 --- a/src/Authentication.cc +++ b/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 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(); + std::shared_ptr tokenSupplier = this->tokenSupplier; Napi::ThreadSafeFunction callback = Napi::ThreadSafeFunction::New( obj.Env(), obj.Get(PARAM_TOKEN).As(), "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) { diff --git a/src/Authentication.h b/src/Authentication.h index 2a507dd..7e51a71 100644 --- a/src/Authentication.h +++ b/src/Authentication.h @@ -20,6 +20,7 @@ #ifndef AUTH_H #define AUTH_H +#include #include #include #include "TokenSupplier.h" @@ -35,7 +36,7 @@ class Authentication : public Napi::ObjectWrap { static Napi::FunctionReference constructor; pulsar_authentication_t *cAuthentication; - TokenSupplierCallback *tokenSupplier; + std::shared_ptr tokenSupplier; }; #endif diff --git a/src/Client.cc b/src/Client.cc index 581b62c..d11f920 100644 --- a/src/Client.cc +++ b/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 Client::logCallback; +std::mutex Client::logMutex; struct _pulsar_client_configuration { pulsar::ClientConfiguration conf; @@ -259,27 +260,34 @@ static std::unique_ptr 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(), "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 next; + if (jsFunction.IsFunction()) { + next = std::make_shared(); + next->callback = Napi::ThreadSafeFunction::New(env, jsFunction.As(), "Pulsar Logging", 0, + 1, [next](Napi::Env) { + std::lock_guard lock(logMutex); + if (logCallback == next) { + logCallback.reset(); + } + }); + next->callback.Unref(env); } + + std::lock_guard 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(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(), "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 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) { diff --git a/src/Client.h b/src/Client.h index 6ea9a15..d80fce9 100644 --- a/src/Client.h +++ b/src/Client.h @@ -20,6 +20,8 @@ #ifndef CLIENT_H #define CLIENT_H +#include +#include #include #include #include @@ -50,13 +52,15 @@ class Client : public Napi::ObjectWrap { ~Client(); private: - static LogCallback *logCallback; + static std::shared_ptr logCallback; + static std::mutex logMutex; static Napi::FunctionReference constructor; std::shared_ptr cClient; std::shared_ptr cClientConfig; pulsar_logger_level_t logLevel = pulsar_logger_level_t::pulsar_INFO; std::vector 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); diff --git a/src/TokenSupplier.h b/src/TokenSupplier.h index 61132a5..d641aca 100644 --- a/src/TokenSupplier.h +++ b/src/TokenSupplier.h @@ -24,6 +24,7 @@ struct TokenSupplierCallback { Napi::ThreadSafeFunction callback; + bool finalized = false; TokenSupplierCallback() {} }; diff --git a/tests/process_exit.test.js b/tests/process_exit.test.js new file mode 100644 index 0000000..d72ee24 --- /dev/null +++ b/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); +}); diff --git a/tests/process_exit/runner.js b/tests/process_exit/runner.js new file mode 100644 index 0000000..7148a08 --- /dev/null +++ b/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); diff --git a/tests/process_exit/token_supplier.js b/tests/process_exit/token_supplier.js new file mode 100644 index 0000000..6c5bbf4 --- /dev/null +++ b/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(); +};