tsfn: support direct calls to underlying napi_tsfn

support direct calls to underlying napi_tsfn

Fixes: #556
PR-URL: #58
Reviewed-By: Michael Dawson <[email protected]>
Reviewed-By: Chengzhong Wu <[email protected]>
Reviewed-By: Gabriel Schulhof <[email protected]>

Author: KevinEady

doc/threadsafe_function.md

@@ -58,7 +58,10 @@ Napi::ThreadSafeFunction::ThreadSafeFunction(napi_threadsafe_function tsfn);
 - `tsfn`: The `napi_threadsafe_function` which is a handle for an existing
   thread-safe function.
 
-Returns a non-empty `Napi::ThreadSafeFunction` instance.
+Returns a non-empty `Napi::ThreadSafeFunction` instance. When using this
+constructor, only use the `Blocking(void*)` / `NonBlocking(void*)` overloads;
+the `Callback` and templated `data*` overloads should _not_ be used. See below
+for additional details.
 
 ### New
 
@@ -171,6 +174,9 @@ There are several overloaded implementations of `BlockingCall()` and
 `NonBlockingCall()` for use with optional parameters: skip the optional
 parameter for that specific overload.
 
+**These specific function overloads should only be used on a `ThreadSafeFunction`
+created via `ThreadSafeFunction::New`.**
+
 ```cpp
 napi_status Napi::ThreadSafeFunction::BlockingCall(DataType* data, Callback callback) const
 
@@ -186,6 +192,17 @@ napi_status Napi::ThreadSafeFunction::NonBlockingCall(DataType* data, Callback c
   necessary to call into JavaScript via `MakeCallback()` because N-API runs
   `callback` in a context appropriate for callbacks.
 
+**These specific function overloads should only be used on a `ThreadSafeFunction`
+created via `ThreadSafeFunction(napi_threadsafe_function)`.**
+
+```cpp
+napi_status Napi::ThreadSafeFunction::BlockingCall(void* data) const
+
+napi_status Napi::ThreadSafeFunction::NonBlockingCall(void* data) const
+```
+- `data`: Data to pass to `call_js_cb` specified when creating the thread-safe
+  function via `napi_create_threadsafe_function`.
+
 Returns one of:
 - `napi_ok`: The call was successfully added to the queue.
 - `napi_queue_full`: The queue was full when trying to call in a non-blocking

napi-inl.h

@@ -4043,6 +4043,12 @@ inline napi_status ThreadSafeFunction::BlockingCall() const {
   return CallInternal(nullptr, napi_tsfn_blocking);
 }
 
+template <>
+inline napi_status ThreadSafeFunction::BlockingCall(
+    void* data) const {
+  return napi_call_threadsafe_function(_tsfn, data, napi_tsfn_blocking);
+}
+
 template <typename Callback>
 inline napi_status ThreadSafeFunction::BlockingCall(
     Callback callback) const {
@@ -4062,6 +4068,12 @@ inline napi_status ThreadSafeFunction::NonBlockingCall() const {
   return CallInternal(nullptr, napi_tsfn_nonblocking);
 }
 
+template <>
+inline napi_status ThreadSafeFunction::NonBlockingCall(
+    void* data) const {
+  return napi_call_threadsafe_function(_tsfn, data, napi_tsfn_nonblocking);
+}
+
 template <typename Callback>
 inline napi_status ThreadSafeFunction::NonBlockingCall(
     Callback callback) const {

test/binding.cc

@@ -40,6 +40,7 @@ Object InitObjectDeprecated(Env env);
 #endif // !NODE_ADDON_API_DISABLE_DEPRECATED
 Object InitPromise(Env env);
 #if (NAPI_VERSION > 3)
+Object InitThreadSafeFunctionExistingTsfn(Env env);
 Object InitThreadSafeFunctionPtr(Env env);
 Object InitThreadSafeFunctionSum(Env env);
 Object InitThreadSafeFunctionUnref(Env env);
@@ -91,6 +92,7 @@ Object Init(Env env, Object exports) {
 #endif // !NODE_ADDON_API_DISABLE_DEPRECATED
   exports.Set("promise", InitPromise(env));
 #if (NAPI_VERSION > 3)
+  exports.Set("threadsafe_function_existing_tsfn", InitThreadSafeFunctionExistingTsfn(env));
   exports.Set("threadsafe_function_ptr", InitThreadSafeFunctionPtr(env));
   exports.Set("threadsafe_function_sum", InitThreadSafeFunctionSum(env));
   exports.Set("threadsafe_function_unref", InitThreadSafeFunctionUnref(env));

test/binding.gyp

@@ -36,6 +36,7 @@
         'object/object.cc',
         'object/set_property.cc',
         'promise.cc',
+        'threadsafe_function/threadsafe_function_existing_tsfn.cc',
         'threadsafe_function/threadsafe_function_ptr.cc',
         'threadsafe_function/threadsafe_function_sum.cc',
         'threadsafe_function/threadsafe_function_unref.cc',

test/index.js

@@ -39,6 +39,7 @@ let testModules = [
   'object/object_deprecated',
   'object/set_property',
   'promise',
+  'threadsafe_function/threadsafe_function_existing_tsfn',
   'threadsafe_function/threadsafe_function_ptr',
   'threadsafe_function/threadsafe_function_sum',
   'threadsafe_function/threadsafe_function_unref',
@@ -68,6 +69,7 @@ if (napiVersion < 3) {
 
 if (napiVersion < 4) {
   testModules.splice(testModules.indexOf('asyncprogressworker'), 1);
+  testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_existing_tsfn'), 1);
   testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_ptr'), 1);
   testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_sum'), 1);
   testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_unref'), 1);

test/threadsafe_function/threadsafe_function_existing_tsfn.cc

@@ -0,0 +1,112 @@
+#include "napi.h"
+#include <cstdlib>
+
+#if (NAPI_VERSION > 3)
+
+using namespace Napi;
+
+namespace {
+
+struct TestContext {
+  TestContext(Promise::Deferred &&deferred)
+      : deferred(std::move(deferred)), callData(nullptr){};
+
+  napi_threadsafe_function tsfn;
+  Promise::Deferred deferred;
+  double *callData;
+
+  ~TestContext() {
+    if (callData != nullptr)
+      delete callData;
+  };
+};
+
+void FinalizeCB(napi_env env, void * /*finalizeData */, void *context) {
+  TestContext *testContext = static_cast<TestContext *>(context);
+  if (testContext->callData != nullptr) {
+    testContext->deferred.Resolve(Number::New(env, *testContext->callData));
+  } else {
+    testContext->deferred.Resolve(Napi::Env(env).Undefined());
+  }
+  delete testContext;
+}
+
+void CallJSWithData(napi_env env, napi_value /* callback */, void *context,
+                    void *data) {
+  TestContext *testContext = static_cast<TestContext *>(context);
+  testContext->callData = static_cast<double *>(data);
+
+  napi_status status =
+      napi_release_threadsafe_function(testContext->tsfn, napi_tsfn_release);
+
+  NAPI_THROW_IF_FAILED_VOID(env, status);
+}
+
+void CallJSNoData(napi_env env, napi_value /* callback */, void *context,
+                  void * /*data*/) {
+  TestContext *testContext = static_cast<TestContext *>(context);
+  testContext->callData = nullptr;
+
+  napi_status status =
+      napi_release_threadsafe_function(testContext->tsfn, napi_tsfn_release);
+
+  NAPI_THROW_IF_FAILED_VOID(env, status);
+}
+
+static Value TestCall(const CallbackInfo &info) {
+  Napi::Env env = info.Env();
+  bool isBlocking = false;
+  bool hasData = false;
+  if (info.Length() > 0) {
+    Object opts = info[0].As<Object>();
+    if (opts.Has("blocking")) {
+      isBlocking = opts.Get("blocking").ToBoolean();
+    }
+    if (opts.Has("data")) {
+      hasData = opts.Get("data").ToBoolean();
+    }
+  }
+
+  // Allow optional callback passed from JS. Useful for testing.
+  Function cb = Function::New(env, [](const CallbackInfo & /*info*/) {});
+
+  TestContext *testContext = new TestContext(Napi::Promise::Deferred(env));
+
+  napi_status status = napi_create_threadsafe_function(
+      env, cb, Object::New(env), String::New(env, "Test"), 0, 1,
+      nullptr, /*finalize data*/
+      FinalizeCB, testContext, hasData ? CallJSWithData : CallJSNoData,
+      &testContext->tsfn);
+
+  NAPI_THROW_IF_FAILED(env, status, Value());
+
+  ThreadSafeFunction wrapped = ThreadSafeFunction(testContext->tsfn);
+
+  // Test the four napi_threadsafe_function direct-accessing calls
+  if (isBlocking) {
+    if (hasData) {
+      wrapped.BlockingCall(static_cast<void *>(new double(std::rand())));
+    } else {
+      wrapped.BlockingCall(static_cast<void *>(nullptr));
+    }
+  } else {
+    if (hasData) {
+      wrapped.NonBlockingCall(static_cast<void *>(new double(std::rand())));
+    } else {
+      wrapped.NonBlockingCall(static_cast<void *>(nullptr));
+    }
+  }
+
+  return testContext->deferred.Promise();
+}
+
+} // namespace
+
+Object InitThreadSafeFunctionExistingTsfn(Env env) {
+  Object exports = Object::New(env);
+  exports["testCall"] = Function::New(env, TestCall);
+
+  return exports;
+}
+
+#endif

test/threadsafe_function/threadsafe_function_existing_tsfn.js

@@ -0,0 +1,19 @@
+'use strict';
+
+const assert = require('assert');
+
+const buildType = process.config.target_defaults.default_configuration;
+
+module.exports = Promise.all([
+  test(require(`../build/${buildType}/binding.node`)),
+  test(require(`../build/${buildType}/binding_noexcept.node`))
+]);
+
+async function test(binding) {
+  const testCall = binding.threadsafe_function_existing_tsfn.testCall;
+  
+  assert(typeof await testCall({ blocking: true,  data: true  }) === "number");
+  assert(typeof await testCall({ blocking: true,  data: false }) === "undefined");
+  assert(typeof await testCall({ blocking: false, data: true  }) === "number");
+  assert(typeof await testCall({ blocking: false, data: false }) === "undefined");
+}