C++ 네이티브 코드에서 Nodejs 비동기 함수 호출

Nodejs 비동기 프로그래밍은 효율적이고 사용자 친화적인 Node.js 애플리케이션을 구축하는 데 필수적인 부분입니다. 그러나 Node.js 애플리케이션용 Node 애드온에서 비동기 코드를 적절하게 처리하는 것은 어려울 수 있습니다. 이 블로그에서는 C++ 네이티브 코드에서 Nodejs 비동기 함수를 호출하는 방법에 대해 설명합니다.

 

---

우리는 Pulsar 소비자에서 트리거 된 이벤트가 발생할 때 Node.js 함수를 호출하기 위해 C++ 네이티브 코드에서 Pulsar 소비자 메시지 수신기를 사용하려고 시도하고 있습니다. 이 메시지 리스너의 인터페이스는 Pulsar C 클라이언트 라이브러리의 일부이며 이 리스너를 연결하는 C++ 함수를 설정했습니다. Node.js 함수인 사용자 함수는 해당 C++ 함수에서 호출됩니다. 

 

---

 

다음은 이 사용 사례에 대한 단순화된 워크플로우입니다.

1. Pulsar 소비자는 트리거 이벤트를 수신합니다.
2. Pulsar 소비자는 C++ 네이티브 코드로 작성된 메시지 수신기에 알립니다.
3. 메시지 리스너는 JS 함수를 호출합니다.
4. 사용자 JS 함수는 이벤트를 비동기적으로 처리합니다.
5. 메시지 리스너는 JS 함수의 완료를 기다립니다.

사용자 JS 기능이 동기화되면 모든 것이 예상대로 작동합니다. 그러나 다음과 같은 비동기 함수를 사용하려고 할 때:

listener: async (message, messageConsumer) => {
          await new Promise((resolve) => setTimeout(resolve, 10));
          consumer1Recv += 1;
          await messageConsumer.acknowledge(message);
        },

메시지 리스너는 사용자 JS 함수의 완료를 기다리지 않습니다. 이는 비동기 JS 함수가 약속 개체를 반환하고 논리를 비동기적으로 실행하기 때문입니다. 결과적으로 5단계는 실제로 사용자 JS 함수의 완료가 아닌 약속 객체를 기다리고 있습니다.

이 블로그 게시물에서는 이 문제를 해결하는 방법에 대해 설명합니다. 먼저 사용자 JS 함수가 완료되는 시기를 확인한 다음 사용자 JS 함수가 완전히 완료될 때까지 기다리도록 C++ 네이티브 코드에서 가드를 설정하는 방법을 알아내야 합니다. 또한 오류를 올바르게 처리하는 방법에 대해서도 이야기합니다.

 

JS 함수 호출을 위한 워크플로

JS 함수를 호출하기 위한 워크플로의 개요부터 살펴보겠습니다. 사용자가 JS 콜백 함수를 C++ 네이티브 코드에 전달할 수 있도록 구성 메서드를 추가했습니다.

Napi::Function jsFunction;
Napi::ThreadSafeFunction callback = Napi::ThreadSafeFunction::New(
        consumerConfig.Env(), jsFunction, "Listener Callback", 1,
        1, (void *)NULL, FinalizeListenerCallback, listener);

가 사용자 JS 코드에서 전달되었다고 가정하면 for this jsFunction를 만듭니다 . 이 유형은 스레드가 대신 JavaScript 함수를 호출하기 위해 Nodejs의 기본 스레드와 통신할 수 있도록 스레드에 API를 제공합니다. ThreadSafeFunctionjsFunctionNapi::ThreadSafeFunctionNapi::ThreadSafeFunction::New

트리거 된 이벤트가 발생하면 callback메시지 리스너 구현에서 위에서 생성한 이벤트를 호출합니다.

 

callback.BlockingCall(dataPtr, MessageListenerProxy);
callback.Release();

를 호출하면 BlockingCallNode.js 스레드 내부의 대기열에서 JS 함수의 실제 호출이 가능해질 때까지 현재 스레드가 차단됩니다.  MessageListenerProxy내부 JS 함수를 호출할 함수에 대한 포인터입니다.

다음은 단순화된 구현입니다.

void MessageListenerProxy(Napi::Env env, Napi::Function jsCallback, MessageListenerProxyData *data) {
  Napi::Object msg = Message::NewInstance({}, data->cMessage);
  jsCallback.Call({ msg }});
}

위는 작업 흐름의 개요입니다. 문제는 에서 발생합니다 jsCallback.Call({ msg }});. jsCallback가 비동기 함수인 경우 jsCallback.Call({ msg }})사용자 함수가 완료될 때까지 기다리지 않고 즉시 반환됩니다.

Nodejs를 사용하여 Promise사용자 기능이 끝나는 지점에 도달

다행스럽게도 는 함수가 비동기인 경우 처리할 jsCallback.CallNode.js 객체를 반환합니다.JavaScript에서 했던 것과 같은 방식으로 Promise처리할 수 있습니다.라는Promise내부 메서드가 있습니다.에 대한 콜백을 등록하는 데 사용할 수 있습니다. PromisethenPromise

달성 방법은 다음과 같습니다.

Napi::Value ret = jsCallback.Call({msg});
if (ret.IsPromise()) {
  Napi::Promise promise = ret.As<Napi::Promise>();
  Napi::Value thenValue = promise.Get("then");
  Napi::Function then = thenValue.As<Napi::Function>();
  Napi::Function callback =
      Napi::Function::New(env, [](const Napi::CallbackInfo &info) { 
          // the point where the user function is finished
        });
  then.Call(promise, {callback});
}

Promise동기 함수의 경우도 처리하려면 반환 값이 a인지 여부를 확인해야 합니다.Napi::Function콜백으로 새로 만듭니다.그런 다음에 등록합니다 Promise.then().

std::promise결과를 기다리는 데 사용

이제 std::promise가드를 설정하는 데 사용할 수 있습니다. 다음은 가드를 설정하는 간단한 샘플 코드입니다.

std::promise<void> promise;
std::future<void> future = promise.get_future();
Napi::Function callback =
      Napi::Function::New(env, [&promise](const Napi::CallbackInfo &info) { 
          promise.set_value();
        });
then.Call(promise, {callback});

// Will be blocked until the jsCallback is finished.
future.wait();

이제 비동기 JS 함수의 결과를 기다릴 수 있을 것 같습니다. 그러나 이 코드도 올바르지 않습니다. Node.js의 전체 메인 스레드를 차단합니다! 이 코드를 실행하면 전체 프로그램이 차단됩니다.

Node.js 스레드에서 기다리지 마세요

위의 문제를 해결하려면 워크플로 전체에서 스레드 콘텍스트 변경을 이해해야 합니다.

메시지 리스너는 Pulsar C 클라이언트 라이브러리 내부의 리스너 스레드 풀에서 오는 스레드에서 실행 중입니다. ThreadSafeFunction그런 다음 호출 콘텍스트를 보다 스레드 안전한 콘텍스트인 Node.js 기본 스레드 콘텍스트로 변경하는 데 사용합니다 . 우리는 Node.js 메인 스레드에서 내부 JS 함수를 호출하고 있습니다. std::future따라서 Node.js 메인 스레드 외부에서 기다려야 합니다.

위에서 설정한 가드를 메시지 리스너 구현으로 이동해야 합니다.

std::promise<void> promise;
std::future<void> future = promise.get_future();
MessageListenerProxyData *dataPtr =
    new MessageListenerProxyData(cMessage, [&promise]() { promise.set_value(); });
listenerCallback->callback.BlockingCall(dataPtr, MessageListenerProxy);
listenerCallback->callback.Release();

future.wait();
delete dataPtr;

MessageListenerProxyData에 람다 콜백을 전달하는 데 사용합니다 MessageListenerProxy. 그리고 에서 MessageListenerProxy이 콜백을 간단히 호출할 수 있습니다.

void MessageListenerProxy(Napi::Env env, Napi::Function jsCallback, MessageListenerProxyData *data) {
  Napi::Object msg = Message::NewInstance({}, data->cMessage);
  Napi::Value ret = jsCallback.Call({msg, consumer->Value()});
  if (ret.IsPromise()) {
    Napi::Promise promise = ret.As<Napi::Promise>();
    Napi::Value thenValue = promise.Get("then");
    if (thenValue.IsFunction()) {
      Napi::Function then = thenValue.As<Napi::Function>();
      Napi::Function callback =
          Napi::Function::New(env, [data](const Napi::CallbackInfo &info) { data->callback(); });
      then.Call(promise, {callback});
      return;
    }
  }
  data->callback();
}

jsCallback동기 함수인 경우에도 가드를 설정합니다..BlockingCall_ data->callback()따라서 동기식일 때 여전히 실행해야 합니다.

오류 처리

promise.catch() 마찬가지로 Nodejs 코드에서 비동기적으로 발생하는 오류를 처리하는 데 사용할 수도 있습니다.

Napi::Function catchFunc = promise.Get("catch").As<Napi::Function>();
ret = catchFunc.Call(promise, {Napi::Function::New(env, [](const Napi::CallbackInfo &info) {
                             Napi::Error error = info[0].As<Napi::Error>();
                             LOG_INFO(error.what())
                             data->callback();
                           })});

여기서 우리는 를 사용하여 유형의 info[0].As<Napi::Error>()첫 번째 매개변수를 얻습니다.오류 메시지를 반환합니다. promise.catch() Errorerror.what()

마지막으로 를 두 번 호출하지 않도록 를 data->callback로 이동하여 코드를 최적화할 수 있습니다. promise.finally() data->callback

promise = ret.As<Napi::Promise>();
Napi::Function finallyFunc = promise.Get("finally").As<Napi::Function>();

finallyFunc.Call(
    promise, {Napi::Function::New(env, [data](const Napi::CallbackInfo &info) { data->callback(); })});

결론적으로 Node.js 애플리케이션용 C++ 애드온에서 비동기 코드를 적절하게 처리하는 것은 까다로울 수 있습니다. 스레드 콘텍스트 변경 사항을 이해하고 Promise 개체 및 가드를 적절하게 관리해야 합니다. ThreadSafeFunction 및 std::promise를 사용하여 애드온이 스레드로부터 안전하고 Node.js 기본 스레드를 차단하지 않도록 할 수 있습니다. C++ 애드온에서 비동기 코드를 적절하게 처리하는 것은 효율적이고 안정적인 Node.js 애플리케이션을 만드는 데 필수적입니다.