edon/napi/
threadsafe_function.rs

1#![allow(clippy::single_component_path_imports)]
2
3use std::convert::Into;
4use std::ffi::CString;
5use std::marker::PhantomData;
6use std::os::raw::c_void;
7use std::ptr::null_mut;
8use std::ptr::{self};
9use std::sync::atomic::AtomicBool;
10use std::sync::atomic::AtomicPtr;
11use std::sync::atomic::Ordering;
12use std::sync::Arc;
13use std::sync::RwLock;
14use std::sync::RwLockWriteGuard;
15use std::sync::Weak;
16
17use libnode_sys;
18
19use crate::napi::bindgen_runtime::FromNapiValue;
20use crate::napi::bindgen_runtime::JsValuesTupleIntoVec;
21use crate::napi::bindgen_runtime::ToNapiValue;
22use crate::napi::bindgen_runtime::TypeName;
23use crate::napi::bindgen_runtime::ValidateNapiValue;
24use crate::napi::check_status;
25use crate::napi::Env;
26use crate::napi::JsError;
27use crate::napi::JsUnknown;
28use crate::napi::Result;
29use crate::napi::Status;
30
31/// ThreadSafeFunction Context object
32/// the `value` is the value passed to `call` method
33pub struct ThreadSafeCallContext<T: 'static> {
34  pub env: Env,
35  pub value: T,
36}
37
38#[repr(u8)]
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
40pub enum ThreadsafeFunctionCallMode {
41  NonBlocking,
42  Blocking,
43}
44
45impl From<ThreadsafeFunctionCallMode> for libnode_sys::napi_threadsafe_function_call_mode {
46  fn from(value: ThreadsafeFunctionCallMode) -> Self {
47    match value {
48      ThreadsafeFunctionCallMode::Blocking => libnode_sys::ThreadsafeFunctionCallMode::blocking,
49      ThreadsafeFunctionCallMode::NonBlocking => {
50        libnode_sys::ThreadsafeFunctionCallMode::nonblocking
51      }
52    }
53  }
54}
55
56type_level_enum! {
57  /// Type-level `enum` to express how to feed [`ThreadsafeFunction`] errors to
58  /// the inner [`JsFunction`].
59  ///
60  /// ### Context
61  ///
62  /// For callbacks that expect a `Result`-like kind of input, the convention is
63  /// to have the callback take an `error` parameter as its first parameter.
64  ///
65  /// This way receiving a `Result<Args…>` can be modelled as follows:
66  ///
67  ///   - In case of `Err(error)`, feed that `error` entity as the first parameter
68  ///     of the callback;
69  ///
70  ///   - Otherwise (in case of `Ok(_)`), feed `null` instead.
71  ///
72  /// In pseudo-code:
73  ///
74  /// ```rust,ignore
75  /// match result_args {
76  ///     Ok(args) => {
77  ///         let js_null = /* … */;
78  ///         callback.call(
79  ///             // this
80  ///             None,
81  ///             // args…
82  ///             &iter::once(js_null).chain(args).collect::<Vec<_>>(),
83  ///         )
84  ///     },
85  ///     Err(err) => callback.call(None, &[JsError::from(err)]),
86  /// }
87  /// ```
88  ///
89  /// **Note that the `Err` case can stem from a failed conversion from native
90  /// values to js values when calling the callback!**
91  ///
92  /// That's why:
93  ///
94  /// > **[This][`ErrorStrategy::CalleeHandled`] is the default error strategy**.
95  ///
96  /// In order to opt-out of it, [`ThreadsafeFunction`] has an optional second
97  /// generic parameter (of "kind" [`ErrorStrategy::T`]) that defines whether
98  /// this behavior ([`ErrorStrategy::CalleeHandled`]) or a non-`Result` one
99  /// ([`ErrorStrategy::Fatal`]) is desired.
100  pub enum ErrorStrategy {
101    /// Input errors (including conversion errors) are left for the callee to
102    /// handle:
103    ///
104    /// The callee receives an extra `error` parameter (the first one), which is
105    /// `null` if no error occurred, and the error payload otherwise.
106    CalleeHandled,
107
108    /// Input errors (including conversion errors) are deemed fatal:
109    ///
110    /// they can thus cause a `panic!` or abort the process.
111    ///
112    /// The callee thus is not expected to have to deal with [that extra `error`
113    /// parameter][CalleeHandled], which is thus not added.
114    Fatal,
115  }
116}
117
118struct ThreadsafeFunctionHandle {
119  raw: AtomicPtr<libnode_sys::napi_threadsafe_function__>,
120  aborted: RwLock<bool>,
121  referred: AtomicBool,
122}
123
124impl ThreadsafeFunctionHandle {
125  /// create a Arc to hold the `ThreadsafeFunctionHandle`
126  fn new(raw: libnode_sys::napi_threadsafe_function) -> Arc<Self> {
127    Arc::new(Self {
128      raw: AtomicPtr::new(raw),
129      aborted: RwLock::new(false),
130      referred: AtomicBool::new(true),
131    })
132  }
133
134  /// Lock `aborted` with read access, call `f` with the value of `aborted`, then unlock it
135  fn with_read_aborted<RT, F>(
136    &self,
137    f: F,
138  ) -> RT
139  where
140    F: FnOnce(bool) -> RT,
141  {
142    let aborted_guard = self
143      .aborted
144      .read()
145      .expect("Threadsafe Function aborted lock failed");
146    f(*aborted_guard)
147  }
148
149  /// Lock `aborted` with write access, call `f` with the `RwLockWriteGuard`, then unlock it
150  fn with_write_aborted<RT, F>(
151    &self,
152    f: F,
153  ) -> RT
154  where
155    F: FnOnce(RwLockWriteGuard<bool>) -> RT,
156  {
157    let aborted_guard = self
158      .aborted
159      .write()
160      .expect("Threadsafe Function aborted lock failed");
161    f(aborted_guard)
162  }
163
164  #[allow(clippy::arc_with_non_send_sync)]
165  fn null() -> Arc<Self> {
166    Self::new(null_mut())
167  }
168
169  fn get_raw(&self) -> libnode_sys::napi_threadsafe_function {
170    self.raw.load(Ordering::SeqCst)
171  }
172
173  fn set_raw(
174    &self,
175    raw: libnode_sys::napi_threadsafe_function,
176  ) {
177    self.raw.store(raw, Ordering::SeqCst)
178  }
179}
180
181impl Drop for ThreadsafeFunctionHandle {
182  fn drop(&mut self) {
183    self.with_read_aborted(|aborted| {
184      if !aborted {
185        let release_status = unsafe {
186          libnode_sys::napi_release_threadsafe_function(
187            self.get_raw(),
188            libnode_sys::ThreadsafeFunctionReleaseMode::release,
189          )
190        };
191        assert!(
192          release_status == libnode_sys::Status::napi_ok,
193          "Threadsafe Function release failed {}",
194          Status::from(release_status)
195        );
196      }
197    })
198  }
199}
200
201#[repr(u8)]
202enum ThreadsafeFunctionCallVariant {
203  Direct,
204  WithCallback,
205}
206
207struct ThreadsafeFunctionCallJsBackData<T> {
208  data: T,
209  call_variant: ThreadsafeFunctionCallVariant,
210  callback: Box<dyn FnOnce(Result<JsUnknown>) -> Result<()>>,
211}
212
213/// Communicate with the addon's main thread by invoking a JavaScript function from other threads.
214///
215/// ## Example
216/// An example of using `ThreadsafeFunction`:
217///
218/// ```rust
219/// #[macro_use]
220/// extern crate napi_derive;
221///
222/// use std::thread;
223///
224/// use napi::{
225///     threadsafe_function::{
226///         ThreadSafeCallContext, ThreadsafeFunctionCallMode, ThreadsafeFunctionReleaseMode,
227///     },
228///     CallContext, Error, JsFunction, JsNumber, JsUndefined, Result, Status,
229/// };
230///
231/// #[js_function(1)]
232/// pub fn test_threadsafe_function(ctx: CallContext) -> Result<JsUndefined> {
233///   let func = ctx.get::<JsFunction>(0)?;
234///
235///   let tsfn =
236///       ctx
237///           .env
238///           .create_threadsafe_function(&func, 0, |ctx: ThreadSafeCallContext<Vec<u32>>| {
239///             ctx.value
240///                 .iter()
241///                 .map(|v| ctx.env.create_uint32(*v))
242///                 .collect::<Result<Vec<JsNumber>>>()
243///           })?;
244///
245///   let tsfn_cloned = tsfn.clone();
246///
247///   thread::spawn(move || {
248///       let output: Vec<u32> = vec![0, 1, 2, 3];
249///       // It's okay to call a threadsafe function multiple times.
250///       tsfn.call(Ok(output.clone()), ThreadsafeFunctionCallMode::Blocking);
251///   });
252///
253///   thread::spawn(move || {
254///       let output: Vec<u32> = vec![3, 2, 1, 0];
255///       // It's okay to call a threadsafe function multiple times.
256///       tsfn_cloned.call(Ok(output.clone()), ThreadsafeFunctionCallMode::NonBlocking);
257///   });
258///
259///   ctx.env.get_undefined()
260/// }
261/// ```
262pub struct ThreadsafeFunction<T: 'static, ES: ErrorStrategy::T = ErrorStrategy::CalleeHandled> {
263  handle: Arc<ThreadsafeFunctionHandle>,
264  _phantom: PhantomData<(T, ES)>,
265}
266
267unsafe impl<T: 'static, ES: ErrorStrategy::T> Send for ThreadsafeFunction<T, ES> {}
268unsafe impl<T: 'static, ES: ErrorStrategy::T> Sync for ThreadsafeFunction<T, ES> {}
269
270impl<T: 'static, ES: ErrorStrategy::T> Clone for ThreadsafeFunction<T, ES> {
271  fn clone(&self) -> Self {
272    self.handle.with_read_aborted(|aborted| {
273      if aborted {
274        panic!("ThreadsafeFunction was aborted, can not clone it");
275      };
276
277      Self {
278        handle: self.handle.clone(),
279        _phantom: PhantomData,
280      }
281    })
282  }
283}
284
285impl<T: ToNapiValue> JsValuesTupleIntoVec for T {
286  #[allow(clippy::not_unsafe_ptr_arg_deref)]
287  fn into_vec(
288    self,
289    env: libnode_sys::napi_env,
290  ) -> Result<Vec<libnode_sys::napi_value>> {
291    Ok(vec![unsafe {
292      <T as ToNapiValue>::to_napi_value(env, self)?
293    }])
294  }
295}
296
297macro_rules! impl_js_value_tuple_to_vec {
298  ($($ident:ident),*) => {
299    impl<$($ident: ToNapiValue),*> JsValuesTupleIntoVec for ($($ident,)*) {
300      #[allow(clippy::not_unsafe_ptr_arg_deref)]
301      fn into_vec(self, env: libnode_sys::napi_env) -> Result<Vec<libnode_sys::napi_value>> {
302        #[allow(non_snake_case)]
303        let ($($ident,)*) = self;
304        Ok(vec![$(unsafe { <$ident as ToNapiValue>::to_napi_value(env, $ident)? }),*])
305      }
306    }
307  };
308}
309
310impl_js_value_tuple_to_vec!(A);
311impl_js_value_tuple_to_vec!(A, B);
312impl_js_value_tuple_to_vec!(A, B, C);
313impl_js_value_tuple_to_vec!(A, B, C, D);
314impl_js_value_tuple_to_vec!(A, B, C, D, E);
315impl_js_value_tuple_to_vec!(A, B, C, D, E, F);
316impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G);
317impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H);
318impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I);
319impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J);
320impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K);
321impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L);
322impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M);
323impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N);
324impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O);
325impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P);
326impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q);
327impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R);
328impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S);
329impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T);
330impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U);
331impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V);
332impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W);
333impl_js_value_tuple_to_vec!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X);
334impl_js_value_tuple_to_vec!(
335  A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X, Y
336);
337impl_js_value_tuple_to_vec!(
338  A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q, R, S, T, U, V, W, X, Y, Z
339);
340
341impl<T: JsValuesTupleIntoVec + 'static, ES: ErrorStrategy::T> FromNapiValue
342  for ThreadsafeFunction<T, ES>
343{
344  unsafe fn from_napi_value(
345    env: libnode_sys::napi_env,
346    napi_val: libnode_sys::napi_value,
347  ) -> Result<Self> {
348    Self::create(env, napi_val, 0, |ctx| ctx.value.into_vec(ctx.env.0))
349  }
350}
351
352impl<T: 'static, ES: ErrorStrategy::T> ThreadsafeFunction<T, ES> {
353  /// See [napi_create_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_create_threadsafe_function)
354  /// for more information.
355  pub(crate) fn create<
356    V: ToNapiValue,
357    R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>,
358  >(
359    env: libnode_sys::napi_env,
360    func: libnode_sys::napi_value,
361    max_queue_size: usize,
362    callback: R,
363  ) -> Result<Self> {
364    let mut async_resource_name = ptr::null_mut();
365    let s = "napi_rs_threadsafe_function";
366    let len = s.len();
367    let s = CString::new(s)?;
368    check_status!(unsafe {
369      libnode_sys::napi_create_string_utf8(env, s.as_ptr(), len as isize, &mut async_resource_name)
370    })?;
371
372    let mut raw_tsfn = ptr::null_mut();
373    let callback_ptr = Box::into_raw(Box::new(callback));
374    let handle = ThreadsafeFunctionHandle::null();
375    check_status!(unsafe {
376      libnode_sys::napi_create_threadsafe_function(
377        env,
378        func,
379        ptr::null_mut(),
380        async_resource_name,
381        max_queue_size,
382        1,
383        Arc::downgrade(&handle).into_raw() as *mut c_void, // pass handler to thread_finalize_cb
384        Some(thread_finalize_cb::<T, V, R>),
385        callback_ptr.cast(),
386        Some(call_js_cb::<T, V, R, ES>),
387        &mut raw_tsfn,
388      )
389    })?;
390    handle.set_raw(raw_tsfn);
391
392    Ok(ThreadsafeFunction {
393      handle,
394      _phantom: PhantomData,
395    })
396  }
397
398  /// See [napi_ref_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_ref_threadsafe_function)
399  /// for more information.
400  ///
401  /// "ref" is a keyword so that we use "refer" here.
402  pub fn refer(
403    &mut self,
404    env: &Env,
405  ) -> Result<()> {
406    self.handle.with_read_aborted(|aborted| {
407      if !aborted && !self.handle.referred.load(Ordering::Relaxed) {
408        check_status!(unsafe {
409          libnode_sys::napi_ref_threadsafe_function(env.0, self.handle.get_raw())
410        })?;
411        self.handle.referred.store(true, Ordering::Relaxed);
412      }
413      Ok(())
414    })
415  }
416
417  /// See [napi_unref_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_unref_threadsafe_function)
418  /// for more information.
419  pub fn unref(
420    &mut self,
421    env: &Env,
422  ) -> Result<()> {
423    self.handle.with_read_aborted(|aborted| {
424      if !aborted && self.handle.referred.load(Ordering::Relaxed) {
425        check_status!(unsafe {
426          libnode_sys::napi_unref_threadsafe_function(env.0, self.handle.get_raw())
427        })?;
428        self.handle.referred.store(false, Ordering::Relaxed);
429      }
430      Ok(())
431    })
432  }
433
434  pub fn aborted(&self) -> bool {
435    self.handle.with_read_aborted(|aborted| aborted)
436  }
437
438  pub fn abort(self) -> Result<()> {
439    self.handle.with_write_aborted(|mut aborted_guard| {
440      if !*aborted_guard {
441        check_status!(unsafe {
442          libnode_sys::napi_release_threadsafe_function(
443            self.handle.get_raw(),
444            libnode_sys::ThreadsafeFunctionReleaseMode::abort,
445          )
446        })?;
447        *aborted_guard = true;
448      }
449      Ok(())
450    })
451  }
452
453  /// Get the raw `ThreadSafeFunction` pointer
454  pub fn raw(&self) -> libnode_sys::napi_threadsafe_function {
455    self.handle.get_raw()
456  }
457}
458
459impl<T: 'static> ThreadsafeFunction<T, ErrorStrategy::CalleeHandled> {
460  /// See [napi_call_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_call_threadsafe_function)
461  /// for more information.
462  pub fn call(
463    &self,
464    value: Result<T>,
465    mode: ThreadsafeFunctionCallMode,
466  ) -> Status {
467    self.handle.with_read_aborted(|aborted| {
468      if aborted {
469        return Status::Closing;
470      }
471
472      unsafe {
473        libnode_sys::napi_call_threadsafe_function(
474          self.handle.get_raw(),
475          Box::into_raw(Box::new(value.map(|data| {
476            ThreadsafeFunctionCallJsBackData {
477              data,
478              call_variant: ThreadsafeFunctionCallVariant::Direct,
479              callback: Box::new(|_d: Result<JsUnknown>| Ok(())),
480            }
481          })))
482          .cast(),
483          mode.into(),
484        )
485      }
486      .into()
487    })
488  }
489
490  pub fn call_with_return_value<D: FromNapiValue, F: 'static + FnOnce(D) -> Result<()>>(
491    &self,
492    value: Result<T>,
493    mode: ThreadsafeFunctionCallMode,
494    cb: F,
495  ) -> Status {
496    self.handle.with_read_aborted(|aborted| {
497      if aborted {
498        return Status::Closing;
499      }
500
501      unsafe {
502        libnode_sys::napi_call_threadsafe_function(
503          self.handle.get_raw(),
504          Box::into_raw(Box::new(value.map(|data| {
505            ThreadsafeFunctionCallJsBackData {
506              data,
507              call_variant: ThreadsafeFunctionCallVariant::WithCallback,
508              callback: Box::new(move |d: Result<JsUnknown>| {
509                d.and_then(|d| D::from_napi_value(d.0.env, d.0.value).and_then(cb))
510              }),
511            }
512          })))
513          .cast(),
514          mode.into(),
515        )
516      }
517      .into()
518    })
519  }
520}
521
522impl<T: 'static> ThreadsafeFunction<T, ErrorStrategy::Fatal> {
523  /// See [napi_call_threadsafe_function](https://nodejs.org/api/n-api.html#n_api_napi_call_threadsafe_function)
524  /// for more information.
525  pub fn call(
526    &self,
527    value: T,
528    mode: ThreadsafeFunctionCallMode,
529  ) -> Status {
530    self.handle.with_read_aborted(|aborted| {
531      if aborted {
532        return Status::Closing;
533      }
534
535      unsafe {
536        libnode_sys::napi_call_threadsafe_function(
537          self.handle.get_raw(),
538          Box::into_raw(Box::new(ThreadsafeFunctionCallJsBackData {
539            data: value,
540            call_variant: ThreadsafeFunctionCallVariant::Direct,
541            callback: Box::new(|_d: Result<JsUnknown>| Ok(())),
542          }))
543          .cast(),
544          mode.into(),
545        )
546      }
547      .into()
548    })
549  }
550
551  pub fn call_with_return_value<D: FromNapiValue, F: 'static + FnOnce(D) -> Result<()>>(
552    &self,
553    value: T,
554    mode: ThreadsafeFunctionCallMode,
555    cb: F,
556  ) -> Status {
557    self.handle.with_read_aborted(|aborted| {
558      if aborted {
559        return Status::Closing;
560      }
561
562      unsafe {
563        libnode_sys::napi_call_threadsafe_function(
564          self.handle.get_raw(),
565          Box::into_raw(Box::new(ThreadsafeFunctionCallJsBackData {
566            data: value,
567            call_variant: ThreadsafeFunctionCallVariant::WithCallback,
568            callback: Box::new(move |d: Result<JsUnknown>| {
569              d.and_then(|d| D::from_napi_value(d.0.env, d.0.value).and_then(cb))
570            }),
571          }))
572          .cast(),
573          mode.into(),
574        )
575      }
576      .into()
577    })
578  }
579}
580
581#[allow(unused_variables)]
582unsafe extern "C" fn thread_finalize_cb<T: 'static, V: ToNapiValue, R>(
583  env: libnode_sys::napi_env,
584  finalize_data: *mut c_void,
585  finalize_hint: *mut c_void,
586) where
587  R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>,
588{
589  let handle_option =
590    unsafe { Weak::from_raw(finalize_data.cast::<ThreadsafeFunctionHandle>()).upgrade() };
591
592  if let Some(handle) = handle_option {
593    handle.with_write_aborted(|mut aborted_guard| {
594      if !*aborted_guard {
595        *aborted_guard = true;
596      }
597    });
598  }
599
600  // cleanup
601  drop(unsafe { Box::<R>::from_raw(finalize_hint.cast()) });
602}
603
604unsafe extern "C" fn call_js_cb<T: 'static, V: ToNapiValue, R, ES>(
605  raw_env: libnode_sys::napi_env,
606  js_callback: libnode_sys::napi_value,
607  context: *mut c_void,
608  data: *mut c_void,
609) where
610  R: 'static + Send + FnMut(ThreadSafeCallContext<T>) -> Result<Vec<V>>,
611  ES: ErrorStrategy::T,
612{
613  // env and/or callback can be null when shutting down
614  if raw_env.is_null() || js_callback.is_null() {
615    return;
616  }
617
618  let ctx: &mut R = unsafe { Box::leak(Box::from_raw(context.cast())) };
619  let val = unsafe {
620    match ES::VALUE {
621      ErrorStrategy::CalleeHandled::VALUE => {
622        *Box::<Result<ThreadsafeFunctionCallJsBackData<T>>>::from_raw(data.cast())
623      }
624      ErrorStrategy::Fatal::VALUE => Ok(*Box::<ThreadsafeFunctionCallJsBackData<T>>::from_raw(
625        data.cast(),
626      )),
627    }
628  };
629
630  let mut recv = ptr::null_mut();
631  unsafe { libnode_sys::napi_get_undefined(raw_env, &mut recv) };
632
633  let ret = val.and_then(|v| {
634    (ctx)(ThreadSafeCallContext {
635      env: unsafe { Env::from_raw(raw_env) },
636      value: v.data,
637    })
638    .map(|ret| (ret, v.call_variant, v.callback))
639  });
640
641  // Follow async callback conventions: https://nodejs.org/en/knowledge/errors/what-are-the-error-conventions/
642  // Check if the Result is okay, if so, pass a null as the first (error) argument automatically.
643  // If the Result is an error, pass that as the first argument.
644  let status = match ret {
645    Ok((values, call_variant, callback)) => {
646      let values = values
647        .into_iter()
648        .map(|v| unsafe { ToNapiValue::to_napi_value(raw_env, v) });
649      let args: Result<Vec<libnode_sys::napi_value>> =
650        if ES::VALUE == ErrorStrategy::CalleeHandled::VALUE {
651          let mut js_null = ptr::null_mut();
652          unsafe { libnode_sys::napi_get_null(raw_env, &mut js_null) };
653          ::core::iter::once(Ok(js_null)).chain(values).collect()
654        } else {
655          values.collect()
656        };
657      let mut return_value = ptr::null_mut();
658      let mut status = match args {
659        Ok(args) => unsafe {
660          libnode_sys::napi_call_function(
661            raw_env,
662            recv,
663            js_callback,
664            args.len(),
665            args.as_ptr(),
666            &mut return_value,
667          )
668        },
669        Err(e) => match ES::VALUE {
670          ErrorStrategy::Fatal::VALUE => unsafe {
671            libnode_sys::napi_fatal_exception(raw_env, JsError::from(e).into_value(raw_env))
672          },
673          ErrorStrategy::CalleeHandled::VALUE => unsafe {
674            libnode_sys::napi_call_function(
675              raw_env,
676              recv,
677              js_callback,
678              1,
679              [JsError::from(e).into_value(raw_env)].as_mut_ptr(),
680              &mut return_value,
681            )
682          },
683        },
684      };
685      if let ThreadsafeFunctionCallVariant::WithCallback = call_variant {
686        // throw Error in JavaScript callback
687        let callback_arg = if status == libnode_sys::Status::napi_pending_exception {
688          let mut exception = ptr::null_mut();
689          status =
690            unsafe { libnode_sys::napi_get_and_clear_last_exception(raw_env, &mut exception) };
691          Err(
692            JsUnknown(crate::napi::Value {
693              env: raw_env,
694              value: exception,
695              value_type: crate::napi::ValueType::Unknown,
696            })
697            .into(),
698          )
699        } else {
700          Ok(JsUnknown(crate::napi::Value {
701            env: raw_env,
702            value: return_value,
703            value_type: crate::napi::ValueType::Unknown,
704          }))
705        };
706        if let Err(err) = callback(callback_arg) {
707          let message = format!(
708            "Failed to convert return value in ThreadsafeFunction callback into Rust value: {err}"
709          );
710          let message_length = message.len();
711          let c_message = CString::new(message).unwrap();
712          unsafe {
713            libnode_sys::napi_fatal_error(
714              "threadsafe_function.rs:749\0".as_ptr().cast(),
715              26,
716              c_message.as_ptr(),
717              message_length as isize,
718            )
719          };
720        }
721      }
722      status
723    }
724    Err(e) if ES::VALUE == ErrorStrategy::Fatal::VALUE => unsafe {
725      libnode_sys::napi_fatal_exception(raw_env, JsError::from(e).into_value(raw_env))
726    },
727    Err(e) => unsafe {
728      libnode_sys::napi_call_function(
729        raw_env,
730        recv,
731        js_callback,
732        1,
733        [JsError::from(e).into_value(raw_env)].as_mut_ptr(),
734        ptr::null_mut(),
735      )
736    },
737  };
738  handle_call_js_cb_status(status, raw_env)
739}
740
741fn handle_call_js_cb_status(
742  status: libnode_sys::napi_status,
743  raw_env: libnode_sys::napi_env,
744) {
745  if status == libnode_sys::Status::napi_ok {
746    return;
747  }
748  if status == libnode_sys::Status::napi_pending_exception {
749    let mut error_result = ptr::null_mut();
750    assert_eq!(
751      unsafe { libnode_sys::napi_get_and_clear_last_exception(raw_env, &mut error_result) },
752      libnode_sys::Status::napi_ok
753    );
754
755    // When shutting down, napi_fatal_exception sometimes returns another exception
756    let stat = unsafe { libnode_sys::napi_fatal_exception(raw_env, error_result) };
757    assert!(
758      stat == libnode_sys::Status::napi_ok || stat == libnode_sys::Status::napi_pending_exception
759    );
760  } else {
761    let error_code: Status = status.into();
762    let error_code_string = format!("{error_code:?}");
763    let mut error_code_value = ptr::null_mut();
764    assert_eq!(
765      unsafe {
766        libnode_sys::napi_create_string_utf8(
767          raw_env,
768          error_code_string.as_ptr() as *const _,
769          error_code_string.len() as isize,
770          &mut error_code_value,
771        )
772      },
773      libnode_sys::Status::napi_ok,
774    );
775    let error_msg = "Call JavaScript callback failed in threadsafe function";
776    let mut error_msg_value = ptr::null_mut();
777    assert_eq!(
778      unsafe {
779        libnode_sys::napi_create_string_utf8(
780          raw_env,
781          error_msg.as_ptr() as *const _,
782          error_msg.len() as isize,
783          &mut error_msg_value,
784        )
785      },
786      libnode_sys::Status::napi_ok,
787    );
788    let mut error_value = ptr::null_mut();
789    assert_eq!(
790      unsafe {
791        libnode_sys::napi_create_error(raw_env, error_code_value, error_msg_value, &mut error_value)
792      },
793      libnode_sys::Status::napi_ok,
794    );
795    assert_eq!(
796      unsafe { libnode_sys::napi_fatal_exception(raw_env, error_value) },
797      libnode_sys::Status::napi_ok
798    );
799  }
800}
801
802/// Helper
803macro_rules! type_level_enum {(
804  $( #[doc = $doc:tt] )*
805  $pub:vis
806  enum $EnumName:ident {
807    $(
808      $( #[doc = $doc_variant:tt] )*
809      $Variant:ident
810    ),* $(,)?
811  }
812) => (type_level_enum! { // This requires the macro to be in scope when called.
813  with_docs! {
814    $( #[doc = $doc] )*
815    ///
816    /// ### Type-level `enum`
817    ///
818    /// Until `const_generics` can handle custom `enum`s, this pattern must be
819    /// implemented at the type level.
820    ///
821    /// We thus end up with:
822    ///
823    /// ```rust,ignore
824    /// #[type_level_enum]
825    #[doc = ::core::concat!(
826      " enum ", ::core::stringify!($EnumName), " {",
827    )]
828    $(
829      #[doc = ::core::concat!(
830        "     ", ::core::stringify!($Variant), ",",
831      )]
832    )*
833    #[doc = " }"]
834    /// ```
835    ///
836    #[doc = ::core::concat!(
837      "With [`", ::core::stringify!($EnumName), "::T`](#reexports) \
838      being the type-level \"enum type\":",
839    )]
840    ///
841    /// ```rust,ignore
842    #[doc = ::core::concat!(
843      "<Param: ", ::core::stringify!($EnumName), "::T>"
844    )]
845    /// ```
846  }
847  #[allow(warnings)]
848  $pub mod $EnumName {
849    #[doc(no_inline)]
850    pub use $EnumName as T;
851
852    super::type_level_enum! {
853      with_docs! {
854        #[doc = ::core::concat!(
855          "See [`", ::core::stringify!($EnumName), "`]\
856          [super::", ::core::stringify!($EnumName), "]"
857        )]
858      }
859      pub trait $EnumName : __sealed::$EnumName + ::core::marker::Sized + 'static {
860        const VALUE: __value::$EnumName;
861      }
862    }
863
864    mod __sealed { pub trait $EnumName {} }
865
866    mod __value {
867      #[derive(Debug, PartialEq, Eq)]
868      pub enum $EnumName { $( $Variant ),* }
869    }
870
871    $(
872      $( #[doc = $doc_variant] )*
873      pub enum $Variant {}
874      impl __sealed::$EnumName for $Variant {}
875      impl $EnumName for $Variant {
876        const VALUE: __value::$EnumName = __value::$EnumName::$Variant;
877      }
878      impl $Variant {
879        pub const VALUE: __value::$EnumName = __value::$EnumName::$Variant;
880      }
881    )*
882  }
883});(
884  with_docs! {
885    $( #[doc = $doc:expr] )*
886  }
887  $item:item
888) => (
889  $( #[doc = $doc] )*
890  $item
891)}
892
893use type_level_enum;
894
895pub struct UnknownReturnValue;
896
897impl TypeName for UnknownReturnValue {
898  fn type_name() -> &'static str {
899    "UnknownReturnValue"
900  }
901
902  fn value_type() -> crate::napi::ValueType {
903    crate::napi::ValueType::Unknown
904  }
905}
906
907impl ValidateNapiValue for UnknownReturnValue {}
908
909impl FromNapiValue for UnknownReturnValue {
910  unsafe fn from_napi_value(
911    _env: libnode_sys::napi_env,
912    _napi_val: libnode_sys::napi_value,
913  ) -> Result<Self> {
914    Ok(UnknownReturnValue)
915  }
916}
edon/napi/threadsafe_function.rs

edon/napi/
threadsafe_function.rs
