Function DataSourceThread

test/threadsafe_function/threadsafe_function.cc:44–126

Source thread producing the data

// Source thread producing the data
static void DataSourceThread() {
  ThreadSafeFunctionInfo* info = s_tsfn.GetContext();

  if (info->startSecondary) {
    if (s_tsfn.Acquire() != napi_ok) {
      Error::Fatal("DataSourceThread", "ThreadSafeFunction.Acquire() failed");
    }
    threads[1] = std::thread(SecondaryThread);
  }

  bool queueWasFull = false;
  bool queueWasClosing = false;

  for (int index = ARRAY_LENGTH - 1; index > -1 && !queueWasClosing; index--) {
    napi_status status = napi_generic_failure;

    auto callback = [](Env env, Function jsCallback, int* data) {
      jsCallback.Call({Number::New(env, *data)});
    };

    auto noArgCallback = [](Env env, Function jsCallback) {
      jsCallback.Call({Number::New(env, 42)});
    };

    switch (info->type) {
      case ThreadSafeFunctionInfo::DEFAULT:
        status = s_tsfn.BlockingCall();
        break;
      case ThreadSafeFunctionInfo::BLOCKING:
        status = s_tsfn.BlockingCall(&ints[index], callback);
        break;
      case ThreadSafeFunctionInfo::NON_BLOCKING:
        status = s_tsfn.NonBlockingCall(&ints[index], callback);
        break;
      case ThreadSafeFunctionInfo::NON_BLOCKING_DEFAULT:
        status = s_tsfn.NonBlockingCall();
        break;

      case ThreadSafeFunctionInfo::NON_BLOCKING_SINGLE_ARG:
        status = s_tsfn.NonBlockingCall(noArgCallback);
        break;
    }

    if (info->abort && (info->type == ThreadSafeFunctionInfo::BLOCKING ||
                        info->type == ThreadSafeFunctionInfo::DEFAULT)) {
      // Let's make this thread really busy to give the main thread a chance to
      // abort / close.
      std::unique_lock<std::mutex> lk(info->protect);
      while (!info->closeCalledFromJs) {
        info->signal.wait(lk);
      }
    }

    switch (status) {
      case napi_queue_full:
        queueWasFull = true;
        index++;
        // fall through