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/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(); +};