You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/03/04 10:09:26 UTC
[GitHub] [pulsar-client-node] Matt-Esch opened a new pull request #200: convert worker threads to async functions
Matt-Esch opened a new pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200
I have replaced the use of async workers with direct async functions by utilising a thread safe deferred (promise). The need for this comes from the fact that with workers, calling producer.send in order results in non-deterministic ordering, and awaiting every send would diminish the utility of having a batched mode. I also found that closing the client while worker threads were ready to execute resulted in segfaults, so this change ensures that most operations execute in the order they are called on the js side.
I've also found several places lacking in good memory management discipline and have patched a few leaks. I've opted to use shared pointers with custom destructors to wrap all of the returned pointers, referring the pulsar library code to check who owns these pointers. (I did not change Auth but this could also benefit). I have avoided excessively adding std::move manually as modern compilers are quite good at optimising this themselves, but I would welcome suggestions where appropriate.
I've changed the logger so that it is set globally. Currently the logger is supplied by the client configuration, but this gives the impression that the logger can be set per client. This is not true and the logger can only be set once for the lifetime of the process. The logger change commandeers the logger by default, avoiding unexpected logging to stdout and allows for the js log function to be switched out at runtime.
I have replaced most of the workers with async functions, the only ones I haven't replaced are the ones for which async functions are not exposed to the c api. I've raised an issue for this https://github.com/apache/pulsar/issues/14452, but it should be noted that the library is still prone to segfaulting if the read workers execute after the client close call.
This is my first time contributing to this library and I'm not particularly experienced with C++, but I have written node addons before. I have run the tests and checked that the linting and formatter have been run. If there are concerns around the style (such as the use of auto) I would welcome the feedback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1084389095
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825615679
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
Review comment:
```suggestion
} else {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] equanz commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
equanz commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r832988565
##########
File path: src/Client.cc
##########
@@ -196,33 +236,27 @@ void LogMessage(pulsar_logger_level_t level, const char *file, int line, const c
logCallback->callback.Release();
}
-class ClientCloseWorker : public Napi::AsyncWorker {
- public:
- ClientCloseWorker(const Napi::Promise::Deferred &deferred, pulsar_client_t *cClient)
- : AsyncWorker(Napi::Function::New(deferred.Promise().Env(), [](const Napi::CallbackInfo &info) {})),
- deferred(deferred),
- cClient(cClient) {}
- ~ClientCloseWorker() {}
- void Execute() {
- pulsar_result result = pulsar_client_close(this->cClient);
- if (result != pulsar_result_Ok) SetError(pulsar_result_str(result));
- }
- void OnOK() { this->deferred.Resolve(Env().Null()); }
- void OnError(const Napi::Error &e) {
- this->deferred.Reject(
- Napi::Error::New(Env(), std::string("Failed to close client: ") + e.Message()).Value());
- }
+Napi::Value Client::Close(const Napi::CallbackInfo &info) {
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Client *>(this, deferred);
+ this->Ref();
- private:
- Napi::Promise::Deferred deferred;
- pulsar_client_t *cClient;
-};
+ pulsar_client_close_async(
+ this->cClient.get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Client *> *>(ctx);
Review comment:
I'm not sure, but currently, `deferredContext` is not deleted at the callback. Is it okay?
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,225 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge id: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
void Consumer::NegativeAcknowledge(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
Message *msg = Message::Unwrap(obj);
- pulsar_consumer_negative_acknowledge(this->wrapper->cConsumer, msg->GetCMessage());
+ std::shared_ptr<pulsar_message_t> cMessage = msg->GetCMessage();
+ pulsar_consumer_negative_acknowledge(this->cConsumer.get(), cMessage.get());
}
void Consumer::NegativeAcknowledgeId(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_negative_acknowledge_id(this->wrapper->cConsumer, msgId->GetCMessageId());
+ std::shared_ptr<pulsar_message_id_t> cMessageId = msgId->GetCMessageId();
+ pulsar_consumer_negative_acknowledge_id(this->cConsumer.get(), cMessageId.get());
}
-void Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge cumulatively: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL,
- NULL);
+Napi::Value Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge cumulatively by id: ") +
+ pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
Napi::Value Consumer::IsConnected(const Napi::CallbackInfo &info) {
Napi::Env env = info.Env();
- return Napi::Boolean::New(env, pulsar_consumer_is_connected(this->wrapper->cConsumer));
+ return Napi::Boolean::New(env, pulsar_consumer_is_connected(this->cConsumer.get()));
}
-class ConsumerCloseWorker : public Napi::AsyncWorker {
- public:
- ConsumerCloseWorker(const Napi::Promise::Deferred &deferred, pulsar_consumer_t *cConsumer,
- Consumer *consumer)
- : AsyncWorker(Napi::Function::New(deferred.Promise().Env(), [](const Napi::CallbackInfo &info) {})),
- deferred(deferred),
- cConsumer(cConsumer),
- consumer(consumer) {}
-
- ~ConsumerCloseWorker() {}
- void Execute() {
- pulsar_consumer_pause_message_listener(this->cConsumer);
- pulsar_result result = pulsar_consumer_close(this->cConsumer);
- if (result != pulsar_result_Ok) {
- SetError(pulsar_result_str(result));
- }
- }
- void OnOK() {
- this->consumer->Cleanup();
- this->deferred.Resolve(Env().Null());
- }
- void OnError(const Napi::Error &e) {
- this->deferred.Reject(
- Napi::Error::New(Env(), std::string("Failed to close consumer: ") + e.Message()).Value());
- }
-
- private:
- Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- Consumer *consumer;
-};
-
-class ConsumerUnsubscribeWorker : public Napi::AsyncWorker {
- public:
- ConsumerUnsubscribeWorker(const Napi::Promise::Deferred &deferred, pulsar_consumer_t *cConsumer,
- Consumer *consumer)
- : AsyncWorker(Napi::Function::New(deferred.Promise().Env(), [](const Napi::CallbackInfo &info) {})),
- deferred(deferred),
- cConsumer(cConsumer),
- consumer(consumer) {}
-
- ~ConsumerUnsubscribeWorker() {}
- void Execute() {
- pulsar_consumer_pause_message_listener(this->cConsumer);
- pulsar_result result = pulsar_consumer_unsubscribe(this->cConsumer);
- if (result != pulsar_result_Ok) {
- SetError(pulsar_result_str(result));
- }
- }
- void OnOK() {
- this->consumer->Cleanup();
- this->deferred.Resolve(Env().Null());
- }
- void OnError(const Napi::Error &e) {
- this->deferred.Reject(
- Napi::Error::New(Env(), std::string("Failed to unsubscribe consumer: ") + e.Message()).Value());
- }
-
- private:
- Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- Consumer *consumer;
-};
-
void Consumer::Cleanup() {
- if (this->listener) {
- this->CleanupListener();
+ if (this->listener != nullptr) {
+ pulsar_consumer_pause_message_listener(this->cConsumer.get());
+ this->listener->callback.Release();
+ this->listener = nullptr;
+ this->Unref();
}
}
-void Consumer::CleanupListener() {
- pulsar_consumer_pause_message_listener(this->wrapper->cConsumer);
- this->Unref();
- this->listener->callback.Release();
- this->listener = nullptr;
-}
-
Napi::Value Consumer::Close(const Napi::CallbackInfo &info) {
- Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
- ConsumerCloseWorker *wk = new ConsumerCloseWorker(deferred, this->wrapper->cConsumer, this);
- wk->Queue();
- return deferred.Promise();
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_pause_message_listener(this->cConsumer.get());
+ pulsar_consumer_close_async(
+ this->cConsumer.get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to close consumer: ") + pulsar_result_str(result));
+ } else {
+ self->Cleanup();
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
Napi::Value Consumer::Unsubscribe(const Napi::CallbackInfo &info) {
- Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
- ConsumerUnsubscribeWorker *wk = new ConsumerUnsubscribeWorker(deferred, this->wrapper->cConsumer, this);
- wk->Queue();
- return deferred.Promise();
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_pause_message_listener(this->cConsumer.get());
+ pulsar_consumer_unsubscribe_async(
+ this->cConsumer.get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to unsubscribe consumer: ") + pulsar_result_str(result));
+ } else {
+ self->Cleanup();
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
Consumer::~Consumer() {
- if (this->listener) {
- this->CleanupListener();
+ this->Cleanup();
+ while (this->Unref() != 0) {
+ // If Ref() > 0 then the process is shutting down. We must unref to prevent
+ // double free (once for the env shutdown and once for non-zero refs)
}
Review comment:
If the refcount is already 0, the Reference#Unref might throw an error.
https://github.com/nodejs/node-addon-api/blob/v4.3.0/napi-inl.h#L2940-L2946
https://github.com/nodejs/node/blob/v16.14.2/src/js_native_api_v8.cc#L2549-L2572
Are these lines okay?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r831660197
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
// Maintain reference to consumer, so it won't get garbage collected
// since, when we have a listener, we don't have to maintain reference to consumer (in js code)
this->Ref();
Review comment:
The reference counter should not be incremented if the message listener is not set. This only gets called in the constructor of the Consumer if there is a listener. All this change does is ensure that if someone were to call this in future from somewhere else the references would remain balanced. I will add a check to make sure that the listener being set isn't itself nullptr. The change just makes the code more resilient rather than changing any actual behaviour.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1074688520
I managed to track down the segfault with valgrind and found an appropriate workaround (issue with the node-addon-api double freeing Ref'd instances https://github.com/nodejs/node-addon-api/issues/1153). Valgrind shows no remaining memory access issues on shutdown. This is likely to be an existing latent bug and it wasn't reproducible for me locally but it was reproducible consistently in github actions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825675677
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge id: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
void Consumer::NegativeAcknowledge(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
Message *msg = Message::Unwrap(obj);
- pulsar_consumer_negative_acknowledge(this->wrapper->cConsumer, msg->GetCMessage());
+ std::shared_ptr<pulsar_message_t> cMessage = msg->GetCMessage();
+ pulsar_consumer_negative_acknowledge(this->cConsumer.get(), cMessage.get());
}
void Consumer::NegativeAcknowledgeId(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_negative_acknowledge_id(this->wrapper->cConsumer, msgId->GetCMessageId());
+ std::shared_ptr<pulsar_message_id_t> cMessageId = msgId->GetCMessageId();
+ pulsar_consumer_negative_acknowledge_id(this->cConsumer.get(), cMessageId.get());
}
-void Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge cumulative: ") +
+ pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL,
- NULL);
+Napi::Value Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge cumulative id: ") +
Review comment:
```suggestion
deferred->Reject(std::string("Consumer failed to acknowledge cumulatively by id: ") +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825699449
##########
File path: tests/end_to_end.test.js
##########
@@ -302,7 +302,7 @@ const Pulsar = require('../index.js');
}
await expect(consumer.receive(1000)).rejects.toThrow(
- 'Failed to received message TimeOut',
+ 'Failed to consumer receive message: TimeOut',
Review comment:
```suggestion
'Failed to receive message: TimeOut',
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825664237
##########
File path: src/Consumer.cc
##########
@@ -225,14 +214,17 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
~ConsumerReceiveWorker() {}
void Execute() {
pulsar_result result;
+ pulsar_message_t *rawMessage;
if (timeout > 0) {
- result = pulsar_consumer_receive_with_timeout(this->cConsumer, &(this->cMessage), timeout);
+ result = pulsar_consumer_receive_with_timeout(this->cConsumer.get(), &rawMessage, timeout);
} else {
- result = pulsar_consumer_receive(this->cConsumer, &(this->cMessage));
+ result = pulsar_consumer_receive(this->cConsumer.get(), &rawMessage);
}
if (result != pulsar_result_Ok) {
- SetError(std::string("Failed to received message ") + pulsar_result_str(result));
+ SetError(std::string("Failed to consumer receive message: ") + pulsar_result_str(result));
Review comment:
```suggestion
SetError(std::string("Consumer failed to receive message: ") + pulsar_result_str(result));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825672970
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge id: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
void Consumer::NegativeAcknowledge(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
Message *msg = Message::Unwrap(obj);
- pulsar_consumer_negative_acknowledge(this->wrapper->cConsumer, msg->GetCMessage());
+ std::shared_ptr<pulsar_message_t> cMessage = msg->GetCMessage();
+ pulsar_consumer_negative_acknowledge(this->cConsumer.get(), cMessage.get());
}
void Consumer::NegativeAcknowledgeId(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_negative_acknowledge_id(this->wrapper->cConsumer, msgId->GetCMessageId());
+ std::shared_ptr<pulsar_message_id_t> cMessageId = msgId->GetCMessageId();
+ pulsar_consumer_negative_acknowledge_id(this->cConsumer.get(), cMessageId.get());
}
-void Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge cumulative: ") +
Review comment:
```suggestion
deferred->Reject(std::string("Failed to acknowledge cumulatively: ") +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825664237
##########
File path: src/Consumer.cc
##########
@@ -225,14 +214,17 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
~ConsumerReceiveWorker() {}
void Execute() {
pulsar_result result;
+ pulsar_message_t *rawMessage;
if (timeout > 0) {
- result = pulsar_consumer_receive_with_timeout(this->cConsumer, &(this->cMessage), timeout);
+ result = pulsar_consumer_receive_with_timeout(this->cConsumer.get(), &rawMessage, timeout);
} else {
- result = pulsar_consumer_receive(this->cConsumer, &(this->cMessage));
+ result = pulsar_consumer_receive(this->cConsumer.get(), &rawMessage);
}
if (result != pulsar_result_Ok) {
- SetError(std::string("Failed to received message ") + pulsar_result_str(result));
+ SetError(std::string("Failed to consumer receive message: ") + pulsar_result_str(result));
Review comment:
```suggestion
SetError(std::string("Failed to receive message: ") + pulsar_result_str(result));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r833187982
##########
File path: src/Client.cc
##########
@@ -196,33 +236,27 @@ void LogMessage(pulsar_logger_level_t level, const char *file, int line, const c
logCallback->callback.Release();
}
-class ClientCloseWorker : public Napi::AsyncWorker {
- public:
- ClientCloseWorker(const Napi::Promise::Deferred &deferred, pulsar_client_t *cClient)
- : AsyncWorker(Napi::Function::New(deferred.Promise().Env(), [](const Napi::CallbackInfo &info) {})),
- deferred(deferred),
- cClient(cClient) {}
- ~ClientCloseWorker() {}
- void Execute() {
- pulsar_result result = pulsar_client_close(this->cClient);
- if (result != pulsar_result_Ok) SetError(pulsar_result_str(result));
- }
- void OnOK() { this->deferred.Resolve(Env().Null()); }
- void OnError(const Napi::Error &e) {
- this->deferred.Reject(
- Napi::Error::New(Env(), std::string("Failed to close client: ") + e.Message()).Value());
- }
+Napi::Value Client::Close(const Napi::CallbackInfo &info) {
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Client *>(this, deferred);
+ this->Ref();
- private:
- Napi::Promise::Deferred deferred;
- pulsar_client_t *cClient;
-};
+ pulsar_client_close_async(
+ this->cClient.get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Client *> *>(ctx);
Review comment:
Nope, good catch I thought I had checked all of these. Not a huge fan of having to new/free these but this is a side effect of using the c api I guess.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1084522750
> Is this line no longer needed?
>
> https://github.com/apache/pulsar-client-node/blob/571aa7a5b9598b317400e5a2bfc797cd9120342d/src/Client.h#L40
Yes this is redundant, I've removed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r840192801
##########
File path: src/Client.cc
##########
@@ -40,14 +41,43 @@ static const std::string CFG_TLS_ALLOW_INSECURE = "tlsAllowInsecureConnection";
static const std::string CFG_STATS_INTERVAL = "statsIntervalInSeconds";
static const std::string CFG_LOG = "log";
+LogCallback *Client::logCallback = nullptr;
+
+void Client::SetLogHandler(const Napi::CallbackInfo &info) {
+ Napi::Env env = info.Env();
+ Napi::HandleScope scope(env);
+ Napi::Value jsFunction = info[0];
+
+ if (jsFunction.IsNull()) {
+ if (Client::logCallback != nullptr) {
+ Client::logCallback->callback.Release();
+ delete Client::logCallback;
+ Client::logCallback = nullptr;
+ }
+ } else if (jsFunction.IsFunction()) {
+ Napi::ThreadSafeFunction logFunction =
+ Napi::ThreadSafeFunction::New(env, jsFunction.As<Napi::Function>(), "Pulsar Logging", 0, 1);
+ logFunction.Unref(env);
Review comment:
[question] What is the reason for doing `Unref` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825667640
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
Review comment:
```suggestion
deferred->Reject(std::string("Consumer failed to acknowledge: ") + pulsar_result_str(result));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825628138
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
// Maintain reference to consumer, so it won't get garbage collected
// since, when we have a listener, we don't have to maintain reference to consumer (in js code)
this->Ref();
Review comment:
I think this was purely to make sure that the ref count is balanced. So if we set the listener 10 times, we don't Ref the Consumer more than once, as there is only one corresponding Unref in `Consumer::Cleanup()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825638693
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
// Maintain reference to consumer, so it won't get garbage collected
// since, when we have a listener, we don't have to maintain reference to consumer (in js code)
this->Ref();
Review comment:
I got it. If so, I think the comments above should be modified to match the code.
> // Maintain reference to consumer, so it won't get garbage collected
> // since, when we have a listener, we don't have to maintain reference to consumer (in js code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825672970
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge id: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
void Consumer::NegativeAcknowledge(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
Message *msg = Message::Unwrap(obj);
- pulsar_consumer_negative_acknowledge(this->wrapper->cConsumer, msg->GetCMessage());
+ std::shared_ptr<pulsar_message_t> cMessage = msg->GetCMessage();
+ pulsar_consumer_negative_acknowledge(this->cConsumer.get(), cMessage.get());
}
void Consumer::NegativeAcknowledgeId(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_negative_acknowledge_id(this->wrapper->cConsumer, msgId->GetCMessageId());
+ std::shared_ptr<pulsar_message_id_t> cMessageId = msgId->GetCMessageId();
+ pulsar_consumer_negative_acknowledge_id(this->cConsumer.get(), cMessageId.get());
}
-void Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge cumulative: ") +
Review comment:
```suggestion
deferred->Reject(std::string("Consumer failed to acknowledge cumulatively: ") +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1071795804
FYI the process seems to segfault at the very end of the test run, which is unusual. I've added a segfault handler to try to capture segfaults in tests and print stracktraces using libunwind. This didn't reproduce locally for me so I am going to need to reproduce through github actions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825667640
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
Review comment:
```suggestion
deferred->Reject(std::string("Failed to acknowledge: ") + pulsar_result_str(result));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825685738
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
// Maintain reference to consumer, so it won't get garbage collected
// since, when we have a listener, we don't have to maintain reference to consumer (in js code)
this->Ref();
Review comment:
Is it correct that the reference counter is incremented even if the user does not set the message listener?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r833189580
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,225 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge id: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
void Consumer::NegativeAcknowledge(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
Message *msg = Message::Unwrap(obj);
- pulsar_consumer_negative_acknowledge(this->wrapper->cConsumer, msg->GetCMessage());
+ std::shared_ptr<pulsar_message_t> cMessage = msg->GetCMessage();
+ pulsar_consumer_negative_acknowledge(this->cConsumer.get(), cMessage.get());
}
void Consumer::NegativeAcknowledgeId(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_negative_acknowledge_id(this->wrapper->cConsumer, msgId->GetCMessageId());
+ std::shared_ptr<pulsar_message_id_t> cMessageId = msgId->GetCMessageId();
+ pulsar_consumer_negative_acknowledge_id(this->cConsumer.get(), cMessageId.get());
}
-void Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge cumulatively: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL,
- NULL);
+Napi::Value Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to acknowledge cumulatively by id: ") +
+ pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
Napi::Value Consumer::IsConnected(const Napi::CallbackInfo &info) {
Napi::Env env = info.Env();
- return Napi::Boolean::New(env, pulsar_consumer_is_connected(this->wrapper->cConsumer));
+ return Napi::Boolean::New(env, pulsar_consumer_is_connected(this->cConsumer.get()));
}
-class ConsumerCloseWorker : public Napi::AsyncWorker {
- public:
- ConsumerCloseWorker(const Napi::Promise::Deferred &deferred, pulsar_consumer_t *cConsumer,
- Consumer *consumer)
- : AsyncWorker(Napi::Function::New(deferred.Promise().Env(), [](const Napi::CallbackInfo &info) {})),
- deferred(deferred),
- cConsumer(cConsumer),
- consumer(consumer) {}
-
- ~ConsumerCloseWorker() {}
- void Execute() {
- pulsar_consumer_pause_message_listener(this->cConsumer);
- pulsar_result result = pulsar_consumer_close(this->cConsumer);
- if (result != pulsar_result_Ok) {
- SetError(pulsar_result_str(result));
- }
- }
- void OnOK() {
- this->consumer->Cleanup();
- this->deferred.Resolve(Env().Null());
- }
- void OnError(const Napi::Error &e) {
- this->deferred.Reject(
- Napi::Error::New(Env(), std::string("Failed to close consumer: ") + e.Message()).Value());
- }
-
- private:
- Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- Consumer *consumer;
-};
-
-class ConsumerUnsubscribeWorker : public Napi::AsyncWorker {
- public:
- ConsumerUnsubscribeWorker(const Napi::Promise::Deferred &deferred, pulsar_consumer_t *cConsumer,
- Consumer *consumer)
- : AsyncWorker(Napi::Function::New(deferred.Promise().Env(), [](const Napi::CallbackInfo &info) {})),
- deferred(deferred),
- cConsumer(cConsumer),
- consumer(consumer) {}
-
- ~ConsumerUnsubscribeWorker() {}
- void Execute() {
- pulsar_consumer_pause_message_listener(this->cConsumer);
- pulsar_result result = pulsar_consumer_unsubscribe(this->cConsumer);
- if (result != pulsar_result_Ok) {
- SetError(pulsar_result_str(result));
- }
- }
- void OnOK() {
- this->consumer->Cleanup();
- this->deferred.Resolve(Env().Null());
- }
- void OnError(const Napi::Error &e) {
- this->deferred.Reject(
- Napi::Error::New(Env(), std::string("Failed to unsubscribe consumer: ") + e.Message()).Value());
- }
-
- private:
- Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- Consumer *consumer;
-};
-
void Consumer::Cleanup() {
- if (this->listener) {
- this->CleanupListener();
+ if (this->listener != nullptr) {
+ pulsar_consumer_pause_message_listener(this->cConsumer.get());
+ this->listener->callback.Release();
+ this->listener = nullptr;
+ this->Unref();
}
}
-void Consumer::CleanupListener() {
- pulsar_consumer_pause_message_listener(this->wrapper->cConsumer);
- this->Unref();
- this->listener->callback.Release();
- this->listener = nullptr;
-}
-
Napi::Value Consumer::Close(const Napi::CallbackInfo &info) {
- Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
- ConsumerCloseWorker *wk = new ConsumerCloseWorker(deferred, this->wrapper->cConsumer, this);
- wk->Queue();
- return deferred.Promise();
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_pause_message_listener(this->cConsumer.get());
+ pulsar_consumer_close_async(
+ this->cConsumer.get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to close consumer: ") + pulsar_result_str(result));
+ } else {
+ self->Cleanup();
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
Napi::Value Consumer::Unsubscribe(const Napi::CallbackInfo &info) {
- Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
- ConsumerUnsubscribeWorker *wk = new ConsumerUnsubscribeWorker(deferred, this->wrapper->cConsumer, this);
- wk->Queue();
- return deferred.Promise();
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_pause_message_listener(this->cConsumer.get());
+ pulsar_consumer_unsubscribe_async(
+ this->cConsumer.get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to unsubscribe consumer: ") + pulsar_result_str(result));
+ } else {
+ self->Cleanup();
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
Consumer::~Consumer() {
- if (this->listener) {
- this->CleanupListener();
+ this->Cleanup();
+ while (this->Unref() != 0) {
+ // If Ref() > 0 then the process is shutting down. We must unref to prevent
+ // double free (once for the env shutdown and once for non-zero refs)
}
Review comment:
I'm glad you pointed that out because I think there is some inconsistency between versions. I've added a ref prior to the unref to ensure that the first unref doesn't throw. We really shouldn't need to do this at all but it's clearly a bug in the node api. The first destructor should reset the references. The reset method also frees so that doesn't help here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825615679
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
Review comment:
```suggestion
} else {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825699223
##########
File path: tests/end_to_end.test.js
##########
@@ -106,7 +106,7 @@ const Pulsar = require('../index.js');
consumer.acknowledge(msg2);
await expect(consumer.receive(1000)).rejects.toThrow(
- 'Failed to received message TimeOut',
+ 'Failed to consumer receive message: TimeOut',
Review comment:
```suggestion
'Failed to receive message: TimeOut',
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825670362
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge id: ") + pulsar_result_str(result));
Review comment:
```suggestion
deferred->Reject(std::string("Consumer failed to acknowledge by id: ") + pulsar_result_str(result));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam removed a comment on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam removed a comment on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1084389096
What about this comment?
https://github.com/apache/pulsar-client-node/pull/200/files#issuecomment-1066336745
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam edited a comment on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam edited a comment on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1084389095
What about this comment?
https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1066336745
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1071795804
FYI the process seems to segfault at the very end of the test run, which is unusual. I've added a segfault handler to try to capture segfaults in tests and print stracktraces using libunwind. This didn't reproduce locally for me so I am going to need to reproduce through github actions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] Matt-Esch commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
Matt-Esch commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1082828552
Tests are passing: https://github.com/Matt-Esch/pulsar-client-node/actions/runs/2033472853
Comments have been addressed. Are we ready to accept this pr and merge?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825615311
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
// Maintain reference to consumer, so it won't get garbage collected
// since, when we have a listener, we don't have to maintain reference to consumer (in js code)
this->Ref();
Review comment:
Why increase the reference count when `listener` is `nullptr`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825670362
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge id: ") + pulsar_result_str(result));
Review comment:
```suggestion
deferred->Reject(std::string("Failed to acknowledge by id: ") + pulsar_result_str(result));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#issuecomment-1066336745
Is this line no longer needed?
https://github.com/apache/pulsar-client-node/blob/571aa7a5b9598b317400e5a2bfc797cd9120342d/src/Client.h#L40
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825622206
##########
File path: src/Consumer.cc
##########
@@ -79,144 +81,131 @@ void MessageListener(pulsar_consumer_t *cConsumer, pulsar_message_t *cMessage, v
listenerCallback->callback.Release();
}
-void Consumer::SetCConsumer(std::shared_ptr<CConsumerWrapper> cConsumer) { this->wrapper = cConsumer; }
-void Consumer::SetListenerCallback(ListenerCallback *listener) {
+void Consumer::SetCConsumer(std::shared_ptr<pulsar_consumer_t> cConsumer) { this->cConsumer = cConsumer; }
+void Consumer::SetListenerCallback(MessageListenerCallback *listener) {
if (listener) {
+ // Pass consumer as argument
+ listener->consumer = this;
+ }
+
+ if (this->listener == nullptr) {
// Maintain reference to consumer, so it won't get garbage collected
// since, when we have a listener, we don't have to maintain reference to consumer (in js code)
this->Ref();
Review comment:
FYI: https://github.com/apache/pulsar-client-node/pull/83
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-node] massakam commented on a change in pull request #200: convert worker threads to async functions
Posted by GitBox <gi...@apache.org>.
massakam commented on a change in pull request #200:
URL: https://github.com/apache/pulsar-client-node/pull/200#discussion_r825675677
##########
File path: src/Consumer.cc
##########
@@ -243,160 +235,229 @@ class ConsumerReceiveWorker : public Napi::AsyncWorker {
private:
Napi::Promise::Deferred deferred;
- pulsar_consumer_t *cConsumer;
- pulsar_message_t *cMessage;
+ std::shared_ptr<pulsar_consumer_t> cConsumer;
+ std::shared_ptr<pulsar_message_t> cMessage;
int64_t timeout;
};
Napi::Value Consumer::Receive(const Napi::CallbackInfo &info) {
Napi::Promise::Deferred deferred = Napi::Promise::Deferred::New(info.Env());
if (info[0].IsUndefined()) {
- ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer);
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer);
wk->Queue();
} else {
Napi::Number timeout = info[0].As<Napi::Object>().ToNumber();
- ConsumerReceiveWorker *wk =
- new ConsumerReceiveWorker(deferred, this->wrapper->cConsumer, timeout.Int64Value());
+ ConsumerReceiveWorker *wk = new ConsumerReceiveWorker(deferred, this->cConsumer, timeout.Int64Value());
wk->Queue();
}
return deferred.Promise();
}
-void Consumer::Acknowledge(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::Acknowledge(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge id: ") + pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
void Consumer::NegativeAcknowledge(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
Message *msg = Message::Unwrap(obj);
- pulsar_consumer_negative_acknowledge(this->wrapper->cConsumer, msg->GetCMessage());
+ std::shared_ptr<pulsar_message_t> cMessage = msg->GetCMessage();
+ pulsar_consumer_negative_acknowledge(this->cConsumer.get(), cMessage.get());
}
void Consumer::NegativeAcknowledgeId(const Napi::CallbackInfo &info) {
Napi::Object obj = info[0].As<Napi::Object>();
MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_negative_acknowledge_id(this->wrapper->cConsumer, msgId->GetCMessageId());
+ std::shared_ptr<pulsar_message_id_t> cMessageId = msgId->GetCMessageId();
+ pulsar_consumer_negative_acknowledge_id(this->cConsumer.get(), cMessageId.get());
}
-void Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- Message *msg = Message::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async(this->wrapper->cConsumer, msg->GetCMessage(), NULL, NULL);
+Napi::Value Consumer::AcknowledgeCumulative(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msg = Message::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async(
+ this->cConsumer.get(), msg->GetCMessage().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge cumulative: ") +
+ pulsar_result_str(result));
+ } else {
+ deferred->Resolve(THREADSAFE_DEFERRED_RESOLVER(env.Null()));
+ }
+
+ self->Unref();
+ },
+ ctx);
+
+ return deferred->Promise();
}
-void Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
- Napi::Object obj = info[0].As<Napi::Object>();
- MessageId *msgId = MessageId::Unwrap(obj);
- pulsar_consumer_acknowledge_cumulative_async_id(this->wrapper->cConsumer, msgId->GetCMessageId(), NULL,
- NULL);
+Napi::Value Consumer::AcknowledgeCumulativeId(const Napi::CallbackInfo &info) {
+ auto obj = info[0].As<Napi::Object>();
+ auto *msgId = MessageId::Unwrap(obj);
+ auto deferred = ThreadSafeDeferred::New(Env());
+ auto ctx = new ExtDeferredContext<Consumer *>(this, deferred);
+ this->Ref();
+
+ pulsar_consumer_acknowledge_cumulative_async_id(
+ this->cConsumer.get(), msgId->GetCMessageId().get(),
+ [](pulsar_result result, void *ctx) {
+ auto deferredContext = static_cast<ExtDeferredContext<Consumer *> *>(ctx);
+ auto deferred = deferredContext->deferred;
+ auto self = deferredContext->ref;
+ delete deferredContext;
+
+ if (result != pulsar_result_Ok) {
+ deferred->Reject(std::string("Failed to consumer acknowledge cumulative id: ") +
Review comment:
```suggestion
deferred->Reject(std::string("Failed to acknowledge cumulatively by id: ") +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org