Fix worker_threads support

Signed-off-by: Stephen Belanger <[email protected]>

Author: Qard

binding.gyp

@@ -15,6 +15,7 @@
         'src/connection.cc',
         'src/errors.cc',
         'src/kafka-consumer.cc',
+        'src/per-isolate-data.cc',
         'src/producer.cc',
         'src/topic.cc',
         'src/workers.cc',

package-lock.json

@@ -11,6 +11,7 @@
 #include <vector>
 #include <math.h>
 
+#include "src/per-isolate-data.h"
 #include "src/workers.h"
 #include "src/admin.h"
 
@@ -33,6 +34,10 @@ AdminClient::AdminClient(Conf* gconfig):
     rkqu = NULL;
 }
 
+void AdminClient::delete_instance(void* arg) {
+  delete (static_cast<AdminClient*>(arg));
+}
+
 AdminClient::~AdminClient() {
   Disconnect();
 }
@@ -90,8 +95,6 @@ Baton AdminClient::Disconnect() {
   return Baton(RdKafka::ERR_NO_ERROR);
 }
 
-Nan::Persistent<v8::Function> AdminClient::constructor;
-
 void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::HandleScope scope;
 
@@ -108,7 +111,7 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "disconnect", NodeDisconnect);
   Nan::SetPrototypeMethod(tpl, "setToken", NodeSetToken);
 
-  constructor.Reset(
+  PerIsolateData::For(v8::Isolate::GetCurrent())->AdminClientConstructor().Reset(
     (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
   Nan::Set(exports, Nan::New("AdminClient").ToLocalChecked(),
     tpl->GetFunction(Nan::GetCurrentContext()).ToLocalChecked());
@@ -155,7 +158,8 @@ v8::Local<v8::Object> AdminClient::NewInstance(v8::Local<v8::Value> arg) {
   const unsigned argc = 1;
 
   v8::Local<v8::Value> argv[argc] = { arg };
-  v8::Local<v8::Function> cons = Nan::New<v8::Function>(constructor);
+  v8::Local<v8::Function> cons = Nan::New<v8::Function>(
+    PerIsolateData::For(v8::Isolate::GetCurrent())->AdminClientConstructor());
   v8::Local<v8::Object> instance =
     Nan::NewInstance(cons, argc, argv).ToLocalChecked();
 

src/admin.cc

@@ -12,7 +12,6 @@
 
 #include <nan.h>
 #include <uv.h>
-#include <iostream>
 #include <string>
 #include <vector>
 
@@ -59,6 +58,8 @@ class AdminClient : public Connection {
   explicit AdminClient(Conf* globalConfig);
   ~AdminClient();
 
+  static void delete_instance(void* arg);
+
   rd_kafka_queue_t* rkqu;
 
  private:

src/admin.h

@@ -7,7 +7,6 @@
  * of the MIT license.  See the LICENSE.txt file for details.
  */
 
-#include <iostream>
 #include "src/binding.h"
 
 using NodeKafka::Producer;
@@ -71,4 +70,4 @@ void Init(v8::Local<v8::Object> exports, v8::Local<v8::Value> m_, void* v_) {
       Nan::New(RdKafka::version_str().c_str()).ToLocalChecked());
 }
 
-NODE_MODULE(kafka, Init)
+NODE_MODULE_CONTEXT_AWARE(kafka, Init)

src/binding.cc

@@ -60,14 +60,16 @@ Dispatcher::~Dispatcher() {
     callbacks[i].Reset();
   }
 
+  Deactivate();
+
   uv_mutex_destroy(&async_lock);
 }
 
 // Only run this if we aren't already listening
 void Dispatcher::Activate() {
   if (!async) {
     async = new uv_async_t;
-    uv_async_init(uv_default_loop(), async, AsyncMessage_);
+    uv_async_init(Nan::GetCurrentEventLoop(), async, AsyncMessage_);
 
     async->data = this;
   }

src/callbacks.cc

@@ -7,6 +7,7 @@
  * of the MIT license.  See the LICENSE.txt file for details.
  */
 
+#include <iostream>
 #include <string>
 #include <vector>
 #include <list>
@@ -36,7 +37,14 @@ void Conf::DumpConfig(std::list<std::string> *dump) {
 
 Conf * Conf::create(RdKafka::Conf::ConfType type, v8::Local<v8::Object> object, std::string &errstr) {  // NOLINT
   v8::Local<v8::Context> context = Nan::GetCurrentContext();
-  Conf* rdconf = static_cast<Conf*>(RdKafka::Conf::create(type));
+  Conf* rdconf = new Conf(type);
+
+  if (type == CONF_GLOBAL)
+    rdconf->rk_conf_ = rd_kafka_conf_new();
+  else
+    rdconf->rkt_conf_ = rd_kafka_topic_conf_new();
+
+  // Conf* rdconf = static_cast<Conf*>(RdKafka::Conf::create(type));
 
   v8::MaybeLocal<v8::Array> _property_names = object->GetOwnPropertyNames(
     Nan::GetCurrentContext());
@@ -150,6 +158,10 @@ Conf::~Conf() {
   if (m_rebalance_cb) {
     delete m_rebalance_cb;
   }
+
+  if (m_offset_commit_cb) {
+    delete m_offset_commit_cb;
+  }
 }
 
 }  // namespace NodeKafka

src/config.cc

@@ -11,18 +11,20 @@
 #define SRC_CONFIG_H_
 
 #include <nan.h>
-#include <iostream>
 #include <vector>
 #include <list>
 #include <string>
 
 #include "rdkafkacpp.h"
+#include "rdkafkacpp_int.h"
 #include "src/common.h"
 #include "src/callbacks.h"
 
 namespace NodeKafka {
 
-class Conf : public RdKafka::Conf {
+class Conf : public RdKafka::ConfImpl {
+ private:
+  Conf(RdKafka::Conf::ConfType type) : RdKafka::ConfImpl(type) {}  // NOLINT
  public:
   ~Conf();
 

src/config.h

@@ -54,9 +54,16 @@ Connection::Connection(Conf* gconfig, Conf* tconfig):
     // Perhaps node new methods should report this as an error? But there
     // isn't anything the user can do about it.
     m_gconfig->set("event_cb", &m_event_cb, errstr);
+
+    node::AddEnvironmentCleanupHook(v8::Isolate::GetCurrent(), delete_instance, this);
   }
 
+void Connection::delete_instance(void* arg) {
+  delete (static_cast<Connection*>(arg));
+}
+
 Connection::~Connection() {
+  node::RemoveEnvironmentCleanupHook(v8::Isolate::GetCurrent(), delete_instance, this);
   uv_rwlock_destroy(&m_connection_lock);
 
   if (m_tconfig) {

src/connection.cc

@@ -11,7 +11,6 @@
 #define SRC_CONNECTION_H_
 
 #include <nan.h>
-#include <iostream>
 #include <string>
 #include <vector>
 
@@ -78,6 +77,8 @@ class Connection : public Nan::ObjectWrap {
   Connection(Conf*, Conf*);
   ~Connection();
 
+  static void delete_instance(void* arg);
+
   static Nan::Persistent<v8::Function> constructor;
   static void New(const Nan::FunctionCallbackInfo<v8::Value>& info);
   static Baton rdkafkaErrorToBaton(RdKafka::Error* error);

src/connection.h

@@ -11,7 +11,6 @@
 #define SRC_ERRORS_H_
 
 #include <nan.h>
-#include <iostream>
 #include <string>
 
 #include "rdkafkacpp.h"

src/errors.h

@@ -11,6 +11,7 @@
 #include <vector>
 
 #include "src/kafka-consumer.h"
+#include "src/per-isolate-data.h"
 #include "src/workers.h"
 
 using Nan::FunctionCallbackInfo;
@@ -36,6 +37,10 @@ KafkaConsumer::KafkaConsumer(Conf* gconfig, Conf* tconfig):
     m_consume_loop = nullptr;
   }
 
+void KafkaConsumer::delete_instance(void* arg) {
+  delete (static_cast<KafkaConsumer*>(arg));
+}
+
 KafkaConsumer::~KafkaConsumer() {
   // We only want to run this if it hasn't been run already
   Disconnect();
@@ -558,8 +563,6 @@ std::string KafkaConsumer::RebalanceProtocol() {
   return consumer->rebalance_protocol();
 }
 
-Nan::Persistent<v8::Function> KafkaConsumer::constructor;
-
 void KafkaConsumer::Init(v8::Local<v8::Object> exports) {
   Nan::HandleScope scope;
 
@@ -620,7 +623,8 @@ void KafkaConsumer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "commitSync", NodeCommitSync);
   Nan::SetPrototypeMethod(tpl, "offsetsStore", NodeOffsetsStore);
 
-  constructor.Reset((tpl->GetFunction(Nan::GetCurrentContext()))
+  PerIsolateData::For(v8::Isolate::GetCurrent())->KafkaConsumerConstructor()
+    .Reset((tpl->GetFunction(Nan::GetCurrentContext()))
     .ToLocalChecked());
   Nan::Set(exports, Nan::New("KafkaConsumer").ToLocalChecked(),
     (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
@@ -680,7 +684,8 @@ v8::Local<v8::Object> KafkaConsumer::NewInstance(v8::Local<v8::Value> arg) {
   const unsigned argc = 1;
 
   v8::Local<v8::Value> argv[argc] = { arg };
-  v8::Local<v8::Function> cons = Nan::New<v8::Function>(constructor);
+  v8::Local<v8::Function> cons = Nan::New<v8::Function>(
+    PerIsolateData::For(v8::Isolate::GetCurrent())->KafkaConsumerConstructor());
   v8::Local<v8::Object> instance =
     Nan::NewInstance(cons, argc, argv).ToLocalChecked();
 
@@ -1395,7 +1400,7 @@ NAN_METHOD(KafkaConsumer::NodeDisconnect) {
     // cleanup the async worker
     consumeLoop->WorkComplete();
     consumeLoop->Destroy();
-  
+
     consumer->m_consume_loop = nullptr;
   }
 

src/kafka-consumer.cc

@@ -12,7 +12,6 @@
 
 #include <nan.h>
 #include <uv.h>
-#include <iostream>
 #include <string>
 #include <vector>
 
@@ -95,6 +94,8 @@ class KafkaConsumer : public Connection {
   KafkaConsumer(Conf *, Conf *);
   ~KafkaConsumer();
 
+  static void delete_instance(void* arg);
+
  private:
   static void part_list_print(const std::vector<RdKafka::TopicPartition*>&);
 
@@ -103,7 +104,7 @@ class KafkaConsumer : public Connection {
   bool m_is_subscribed = false;
 
   void* m_consume_loop = nullptr;
-  
+
   // Node methods
   static NAN_METHOD(NodeConnect);
   static NAN_METHOD(NodeSubscribe);

src/kafka-consumer.h

@@ -0,0 +1,57 @@
+/*
+ * node-rdkafka - Node.js wrapper for RdKafka C/C++ library
+ *
+ * Copyright (c) 2016 Blizzard Entertainment
+ *
+ * This software may be modified and distributed under the terms
+ * of the MIT license.  See the LICENSE.txt file for details.
+ */
+
+#include <mutex>
+#include <unordered_map>
+#include <utility>
+
+#include "per-isolate-data.h"
+
+namespace NodeKafka {
+
+static std::unordered_map<v8::Isolate*, PerIsolateData> per_isolate_data_;
+static std::mutex mutex;
+
+PerIsolateData* PerIsolateData::For(v8::Isolate* isolate) {
+  const std::lock_guard<std::mutex> lock(mutex);
+  auto maybe = per_isolate_data_.find(isolate);
+  if (maybe != per_isolate_data_.end()) {
+    return &maybe->second;
+  }
+
+  per_isolate_data_.emplace(std::make_pair(isolate, PerIsolateData()));
+
+  auto pair = per_isolate_data_.find(isolate);
+  auto perIsolateData = &pair->second;
+
+  node::AddEnvironmentCleanupHook(isolate, [](void* data) {
+    const std::lock_guard<std::mutex> lock(mutex);
+    per_isolate_data_.erase(static_cast<v8::Isolate*>(data));
+  }, isolate);
+
+  return perIsolateData;
+}
+
+Nan::Global<v8::Function>& PerIsolateData::AdminClientConstructor() {
+  return admin_client_constructor;
+}
+
+Nan::Global<v8::Function>& PerIsolateData::KafkaConsumerConstructor() {
+  return kafka_consumer_constructor;
+}
+
+Nan::Global<v8::Function>& PerIsolateData::KafkaProducerConstructor() {
+  return kafka_producer_constructor;
+}
+
+Nan::Global<v8::Function>& PerIsolateData::TopicConstructor() {
+  return topic_constructor;
+}
+
+}  // namespace dd

src/per-isolate-data.cc

@@ -0,0 +1,39 @@
+/*
+ * node-rdkafka - Node.js wrapper for RdKafka C/C++ library
+ *
+ * Copyright (c) 2016 Blizzard Entertainment
+ *
+ * This software may be modified and distributed under the terms
+ * of the MIT license.  See the LICENSE.txt file for details.
+ */
+
+#ifndef SRC_PER_ISOLATE_DATA_H_
+#define SRC_PER_ISOLATE_DATA_H_
+
+#include <node.h>
+#include <nan.h>
+#include <v8.h>
+
+namespace NodeKafka {
+
+class PerIsolateData {
+ private:
+  Nan::Global<v8::Function> admin_client_constructor;
+  Nan::Global<v8::Function> kafka_consumer_constructor;
+  Nan::Global<v8::Function> kafka_producer_constructor;
+  Nan::Global<v8::Function> topic_constructor;
+
+  PerIsolateData() {}
+
+ public:
+  static PerIsolateData* For(v8::Isolate* isolate);
+
+  Nan::Global<v8::Function>& AdminClientConstructor();
+  Nan::Global<v8::Function>& KafkaConsumerConstructor();
+  Nan::Global<v8::Function>& KafkaProducerConstructor();
+  Nan::Global<v8::Function>& TopicConstructor();
+};
+
+}  // namespace NodeKafka
+
+#endif  // SRC_PER_ISOLATE_DATA_H_

src/per-isolate-data.h

@@ -10,6 +10,7 @@
 #include <string>
 #include <vector>
 
+#include "src/per-isolate-data.h"
 #include "src/producer.h"
 #include "src/kafka-consumer.h"
 #include "src/workers.h"
@@ -39,12 +40,14 @@ Producer::Producer(Conf* gconfig, Conf* tconfig):
     m_gconfig->set("dr_cb", &m_dr_cb, errstr);
   }
 
+void Producer::delete_instance(void* arg) {
+  delete (static_cast<Producer*>(arg));
+}
+
 Producer::~Producer() {
   Disconnect();
 }
 
-Nan::Persistent<v8::Function> Producer::constructor;
-
 void Producer::Init(v8::Local<v8::Object> exports) {
   Nan::HandleScope scope;
 
@@ -91,7 +94,8 @@ void Producer::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "sendOffsetsToTransaction", NodeSendOffsetsToTransaction);
 
     // connect. disconnect. resume. pause. get meta data
-  constructor.Reset((tpl->GetFunction(Nan::GetCurrentContext()))
+  PerIsolateData::For(v8::Isolate::GetCurrent())->KafkaProducerConstructor()
+    .Reset((tpl->GetFunction(Nan::GetCurrentContext()))
     .ToLocalChecked());
 
   Nan::Set(exports, Nan::New("Producer").ToLocalChecked(),
@@ -153,7 +157,8 @@ v8::Local<v8::Object> Producer::NewInstance(v8::Local<v8::Value> arg) {
   const unsigned argc = 1;
 
   v8::Local<v8::Value> argv[argc] = { arg };
-  v8::Local<v8::Function> cons = Nan::New<v8::Function>(constructor);
+  v8::Local<v8::Function> cons = Nan::New<v8::Function>(
+    PerIsolateData::For(v8::Isolate::GetCurrent())->KafkaProducerConstructor());
   v8::Local<v8::Object> instance =
     Nan::NewInstance(cons, argc, argv).ToLocalChecked();
 

src/producer.cc

@@ -99,6 +99,8 @@ class Producer : public Connection {
   Producer(Conf*, Conf*);
   ~Producer();
 
+  static void delete_instance(void* arg);
+
  private:
   static NAN_METHOD(NodeProduce);
   static NAN_METHOD(NodeSetPartitioner);

src/producer.h

@@ -12,6 +12,7 @@
 
 #include "src/common.h"
 #include "src/connection.h"
+#include "src/per-isolate-data.h"
 #include "src/topic.h"
 
 namespace NodeKafka {
@@ -33,9 +34,15 @@ Topic::Topic(std::string topic_name, RdKafka::Conf* config):
   m_topic_name(topic_name),
   m_config(config) {
   // We probably want to copy the config. May require refactoring if we do not
+  node::AddEnvironmentCleanupHook(v8::Isolate::GetCurrent(), delete_instance, this);
+}
+
+void Topic::delete_instance(void* arg) {
+  delete (static_cast<Topic*>(arg));
 }
 
 Topic::~Topic() {
+  node::RemoveEnvironmentCleanupHook(v8::Isolate::GetCurrent(), delete_instance, this);
   if (m_config) {
     delete m_config;
   }
@@ -74,8 +81,6 @@ Baton offset_store (int32_t partition, int64_t offset) {
 
 */
 
-Nan::Persistent<v8::Function> Topic::constructor;
-
 void Topic::Init(v8::Local<v8::Object> exports) {
   Nan::HandleScope scope;
 
@@ -86,7 +91,8 @@ void Topic::Init(v8::Local<v8::Object> exports) {
   Nan::SetPrototypeMethod(tpl, "name", NodeGetName);
 
   // connect. disconnect. resume. pause. get meta data
-  constructor.Reset((tpl->GetFunction(Nan::GetCurrentContext()))
+  PerIsolateData::For(v8::Isolate::GetCurrent())->TopicConstructor()
+    .Reset((tpl->GetFunction(Nan::GetCurrentContext()))
     .ToLocalChecked());
 
   Nan::Set(exports, Nan::New("Topic").ToLocalChecked(),
@@ -147,7 +153,8 @@ v8::Local<v8::Object> Topic::NewInstance(v8::Local<v8::Value> arg) {
   const unsigned argc = 1;
 
   v8::Local<v8::Value> argv[argc] = { arg };
-  v8::Local<v8::Function> cons = Nan::New<v8::Function>(constructor);
+  v8::Local<v8::Function> cons = Nan::New<v8::Function>(
+    PerIsolateData::For(v8::Isolate::GetCurrent())->TopicConstructor());
   v8::Local<v8::Object> instance =
     Nan::NewInstance(cons, argc, argv).ToLocalChecked();
 

src/topic.cc

@@ -41,6 +41,8 @@ class Topic : public Nan::ObjectWrap {
   Topic(std::string, RdKafka::Conf *);
   ~Topic();
 
+  static void delete_instance(void* arg);
+
   std::string m_topic_name;
   RdKafka::Conf * m_config;
 

src/topic.h

@@ -40,7 +40,7 @@ OffsetsForTimes::OffsetsForTimes(Nan::Callback *callback,
                                  Connection* handle,
                                  std::vector<RdKafka::TopicPartition*> & t,
                                  const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:OffsetsForTimes"),
   m_handle(handle),
   m_topic_partitions(t),
   m_timeout_ms(timeout_ms) {}
@@ -82,7 +82,7 @@ void OffsetsForTimes::HandleErrorCallback() {
 ConnectionMetadata::ConnectionMetadata(
   Nan::Callback *callback, Connection* connection,
   std::string topic, int timeout_ms, bool all_topics) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ConnectionMetadata"),
   m_connection(connection),
   m_topic(topic),
   m_timeout_ms(timeout_ms),
@@ -139,7 +139,7 @@ void ConnectionMetadata::HandleErrorCallback() {
 ConnectionQueryWatermarkOffsets::ConnectionQueryWatermarkOffsets(
   Nan::Callback *callback, Connection* connection,
   std::string topic, int32_t partition, int timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ConnectionQueryWatermarkOffsets"),
   m_connection(connection),
   m_topic(topic),
   m_partition(partition),
@@ -193,7 +193,7 @@ void ConnectionQueryWatermarkOffsets::HandleErrorCallback() {
  */
 
 ProducerConnect::ProducerConnect(Nan::Callback *callback, Producer* producer):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerConnect"),
   producer(producer) {}
 
 ProducerConnect::~ProducerConnect() {}
@@ -240,7 +240,7 @@ void ProducerConnect::HandleErrorCallback() {
 
 ProducerDisconnect::ProducerDisconnect(Nan::Callback *callback,
   Producer* producer):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerDisconnect"),
   producer(producer) {}
 
 ProducerDisconnect::~ProducerDisconnect() {}
@@ -262,6 +262,7 @@ void ProducerDisconnect::HandleOKCallback() {
 }
 
 void ProducerDisconnect::HandleErrorCallback() {
+  Nan::HandleScope scope;
   // This should never run
   assert(0);
 }
@@ -274,7 +275,7 @@ void ProducerDisconnect::HandleErrorCallback() {
 
 ProducerFlush::ProducerFlush(Nan::Callback *callback,
   Producer* producer, int timeout_ms):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerFlush"),
   producer(producer),
   timeout_ms(timeout_ms) {}
 
@@ -312,7 +313,7 @@ void ProducerFlush::HandleOKCallback() {
 
 ProducerInitTransactions::ProducerInitTransactions(Nan::Callback *callback,
   Producer* producer, const int & timeout_ms):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerInitTransactions"),
   producer(producer),
   m_timeout_ms(timeout_ms) {}
 
@@ -357,7 +358,7 @@ void ProducerInitTransactions::HandleErrorCallback() {
  */
 
 ProducerBeginTransaction::ProducerBeginTransaction(Nan::Callback *callback, Producer* producer):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerBeginTransaction"),
   producer(producer) {}
 
 ProducerBeginTransaction::~ProducerBeginTransaction() {}
@@ -403,7 +404,7 @@ void ProducerBeginTransaction::HandleErrorCallback() {
 
 ProducerCommitTransaction::ProducerCommitTransaction(Nan::Callback *callback,
   Producer* producer, const int & timeout_ms):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerCommitTransaction"),
   producer(producer),
   m_timeout_ms(timeout_ms) {}
 
@@ -449,7 +450,7 @@ void ProducerCommitTransaction::HandleErrorCallback() {
 
 ProducerAbortTransaction::ProducerAbortTransaction(Nan::Callback *callback,
   Producer* producer, const int & timeout_ms):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerAbortTransaction"),
   producer(producer),
   m_timeout_ms(timeout_ms) {}
 
@@ -499,7 +500,7 @@ ProducerSendOffsetsToTransaction::ProducerSendOffsetsToTransaction(
     std::vector<RdKafka::TopicPartition *> & t,
     KafkaConsumer* consumer,
     const int & timeout_ms):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:ProducerSendOffsetsToTransaction"),
   producer(producer),
   m_topic_partitions(t),
   consumer(consumer),
@@ -551,7 +552,7 @@ void ProducerSendOffsetsToTransaction::HandleErrorCallback() {
 
 KafkaConsumerConnect::KafkaConsumerConnect(Nan::Callback *callback,
   KafkaConsumer* consumer):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:KafkaConsumerConnect"),
   consumer(consumer) {}
 
 KafkaConsumerConnect::~KafkaConsumerConnect() {}
@@ -601,7 +602,7 @@ void KafkaConsumerConnect::HandleErrorCallback() {
 
 KafkaConsumerDisconnect::KafkaConsumerDisconnect(Nan::Callback *callback,
   KafkaConsumer* consumer):
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:KafkaConsumerDisconnect"),
   consumer(consumer) {}
 
 KafkaConsumerDisconnect::~KafkaConsumerDisconnect() {}
@@ -661,7 +662,7 @@ KafkaConsumerConsumeLoop::KafkaConsumerConsumeLoop(Nan::Callback *callback,
                                      KafkaConsumer* consumer,
                                      const int & timeout_ms,
                                      const int & timeout_sleep_delay_ms) :
-  MessageWorker(callback),
+  MessageWorker(callback, "node-rdkafka:KafkaConsumerConsumeLoop"),
   consumer(consumer),
   m_looping(true),
   m_timeout_ms(timeout_ms),
@@ -798,7 +799,7 @@ KafkaConsumerConsumeNum::KafkaConsumerConsumeNum(Nan::Callback *callback,
                                      KafkaConsumer* consumer,
                                      const uint32_t & num_messages,
                                      const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:KafkaConsumerConsumeNum"),
   m_consumer(consumer),
   m_num_messages(num_messages),
   m_timeout_ms(timeout_ms) {}
@@ -824,7 +825,7 @@ void KafkaConsumerConsumeNum::Execute() {
           if (m_messages.size() > eof_event_count) {
             timeout_ms = 1;
           }
-          
+
           // We will only go into this code path when `enable.partition.eof` is set to true
           // In this case, consumer is also interested in EOF messages, so we return an EOF message
           m_messages.push_back(message);
@@ -872,7 +873,7 @@ void KafkaConsumerConsumeNum::HandleOKCallback() {
     for (std::vector<RdKafka::Message*>::iterator it = m_messages.begin();
         it != m_messages.end(); ++it) {
       RdKafka::Message* message = *it;
-      
+
       switch (message->err()) {
         case RdKafka::ERR_NO_ERROR:
           ++returnArrayIndex;
@@ -890,15 +891,15 @@ void KafkaConsumerConsumeNum::HandleOKCallback() {
             Nan::New<v8::Number>(message->offset()));
           Nan::Set(eofEvent, Nan::New<v8::String>("partition").ToLocalChecked(),
             Nan::New<v8::Number>(message->partition()));
-          
+
           // also store index at which position in the message array this event was emitted
           // this way, we can later emit it at the right point in time
           Nan::Set(eofEvent, Nan::New<v8::String>("messageIndex").ToLocalChecked(),
             Nan::New<v8::Number>(returnArrayIndex));
 
           Nan::Set(eofEventsArray, eofEventsArrayIndex, eofEvent);
       }
-      
+
       delete message;
     }
   }
@@ -940,7 +941,7 @@ void KafkaConsumerConsumeNum::HandleErrorCallback() {
 KafkaConsumerConsume::KafkaConsumerConsume(Nan::Callback *callback,
                                      KafkaConsumer* consumer,
                                      const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:KafkaConsumerConsume"),
   consumer(consumer),
   m_timeout_ms(timeout_ms) {}
 
@@ -998,7 +999,7 @@ KafkaConsumerCommitted::KafkaConsumerCommitted(Nan::Callback *callback,
                                      KafkaConsumer* consumer,
                                      std::vector<RdKafka::TopicPartition*> & t,
                                      const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:KafkaConsumerCommitted"),
   m_consumer(consumer),
   m_topic_partitions(t),
   m_timeout_ms(timeout_ms) {}
@@ -1052,7 +1053,7 @@ KafkaConsumerSeek::KafkaConsumerSeek(Nan::Callback *callback,
                                      KafkaConsumer* consumer,
                                      const RdKafka::TopicPartition * toppar,
                                      const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:KafkaConsumerSeek"),
   m_consumer(consumer),
   m_toppar(toppar),
   m_timeout_ms(timeout_ms) {}
@@ -1108,7 +1109,7 @@ AdminClientCreateTopic::AdminClientCreateTopic(Nan::Callback *callback,
                                                AdminClient* client,
                                                rd_kafka_NewTopic_t* topic,
                                                const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:AdminClientCreateTopic"),
   m_client(client),
   m_topic(topic),
   m_timeout_ms(timeout_ms) {}
@@ -1155,7 +1156,7 @@ AdminClientDeleteTopic::AdminClientDeleteTopic(Nan::Callback *callback,
                                                AdminClient* client,
                                                rd_kafka_DeleteTopic_t* topic,
                                                const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:AdminClientDeleteTopic"),
   m_client(client),
   m_topic(topic),
   m_timeout_ms(timeout_ms) {}
@@ -1203,7 +1204,7 @@ AdminClientCreatePartitions::AdminClientCreatePartitions(
                                          AdminClient* client,
                                          rd_kafka_NewPartitions_t* partitions,
                                          const int & timeout_ms) :
-  ErrorAwareWorker(callback),
+  ErrorAwareWorker(callback, "node-rdkafka:AdminClientCreatePartitions"),
   m_client(client),
   m_partitions(partitions),
   m_timeout_ms(timeout_ms) {}

src/workers.cc

@@ -26,8 +26,9 @@ namespace Workers {
 
 class ErrorAwareWorker : public Nan::AsyncWorker {
  public:
-  explicit ErrorAwareWorker(Nan::Callback* callback_) :
-    Nan::AsyncWorker(callback_),
+  explicit ErrorAwareWorker(Nan::Callback* callback_,
+    const char* resource_name = "node-rdkafka:ErrorAwareWorker") :
+    Nan::AsyncWorker(callback_, resource_name),
     m_baton(RdKafka::ERR_NO_ERROR) {}
   virtual ~ErrorAwareWorker() {}
 
@@ -68,11 +69,12 @@ class ErrorAwareWorker : public Nan::AsyncWorker {
 
 class MessageWorker : public ErrorAwareWorker {
  public:
-  explicit MessageWorker(Nan::Callback* callback_)
-      : ErrorAwareWorker(callback_), m_asyncdata() {
+  explicit MessageWorker(Nan::Callback* callback_,
+    const char* resource_name = "node-rdkafka:ErrorAwareWorker")
+      : ErrorAwareWorker(callback_, resource_name), m_asyncdata() {
     m_async = new uv_async_t;
     uv_async_init(
-      uv_default_loop(),
+      Nan::GetCurrentEventLoop(),
       m_async,
       m_async_message);
     m_async->data = this;

src/workers.h

@@ -0,0 +1,39 @@
+var WorkerThreads = require('worker_threads');
+var Kafka = require('../');
+
+if (WorkerThreads.isMainThread) {
+  var worker = new WorkerThreads.Worker(__filename);
+
+  var timeout = setTimeout(function() {
+    worker.terminate();
+  }, 1000);
+
+  worker.on('message', function(report) {
+    console.log('received message', report);
+  });
+
+  worker.on('exit', function(code) {
+    clearTimeout(timeout);
+    process.exit(code);
+  });
+
+  return;
+}
+
+var stream = Kafka.KafkaConsumer.createReadStream({
+ 	'metadata.broker.list': 'localhost:9092',
+  'client.id': 'kafka-mocha-consumer',
+ 	'group.id': 'kafka-mocha-grp',
+  'allow.auto.create.topics': true,
+  'enable.auto.commit': false,
+  'rebalance_cb': true,
+}, {}, {
+  topics: ['topic']
+});
+
+stream.on('data', function(message) {
+  WorkerThreads.parentPort?.postMessage({ message });
+  stream.consumer.commitMessage(message);
+  stream.consumer.disconnect();
+  stream.close();
+});

test/kafka-consumer-worker.js

@@ -8,7 +8,10 @@
  */
 
 var KafkaConsumer = require('../lib/kafka-consumer');
+var KafkaProducer = require('../lib/producer');
 var t = require('assert');
+var path = require('path');
+var worker_threads = require('worker_threads');
 
 var client;
 var defaultConfig = {
@@ -44,5 +47,37 @@ module.exports = {
       t.deepStrictEqual(client.topicConfig, {});
       t.notEqual(topicConfig, client.topicConfig);
     },
+    'does not crash in a worker': function (cb) {
+      var consumer = new worker_threads.Worker(
+        path.join(__dirname, 'kafka-consumer-worker.js')
+      );
+
+      var timeout = setTimeout(function() {
+        consumer.terminate();
+      }, 1000);
+
+      consumer.on('message', function(msg) {
+        t.strictEqual(msg.value.toString(), 'my message');
+        consumer.terminate();
+      });
+
+      consumer.on('exit', function(code) {
+        clearTimeout(timeout);
+        t.strictEqual(code, 0);
+        cb();
+      });
+
+      consumer.on('online', function() {
+        const stream = Kafka.Producer.createWriteStream({
+          'metadata.broker.list': 'localhost:9092',
+          'client.id': 'kafka-mocha-producer',
+          'dr_cb': true
+        }, {}, {
+          topic: 'topic'
+        });
+
+        stream.write(Buffer.from('my message'));
+      });
+    }
   },
 };

test/kafka-consumer.spec.js

@@ -0,0 +1,36 @@
+var WorkerThreads = require('worker_threads');
+var Kafka = require('../');
+
+if (WorkerThreads.isMainThread) {
+  var worker = new WorkerThreads.Worker(__filename);
+
+  var timeout = setTimeout(function() {
+    worker.terminate();
+  }, 1000);
+
+  worker.on('message', function(report) {
+    console.log('delivery report', report);
+  });
+
+  worker.on('exit', function(code) {
+    clearTimeout(timeout);
+    process.exit(code);
+  });
+
+  return;
+}
+
+const stream = Kafka.Producer.createWriteStream({
+  'metadata.broker.list': 'localhost:9092',
+  'client.id': 'kafka-mocha-producer',
+  'dr_cb': true
+}, {}, {
+  topic: 'topic'
+});
+
+stream.producer.on('delivery-report', function(err, report) {
+  WorkerThreads.parentPort?.postMessage(report);
+  stream.producer.disconnect();
+});
+
+stream.write(Buffer.from('my message'));

test/producer-worker.js

@@ -94,6 +94,26 @@ module.exports = {
         };
 
         client.disconnect(next);
+      },
+      'does not crash in a worker': function (cb) {
+        var producer = new worker_threads.Worker(
+          path.join(__dirname, 'producer-worker.js')
+        );
+
+        var timeout = setTimeout(function() {
+          producer.terminate();
+        }, 1000);
+
+        consumer.on('message', (report) => {
+          t.strictEqual(report.topic, 'topic');
+          producer.terminate();
+        });
+
+        producer.on('exit', function(code) {
+          clearTimeout(timeout);
+          t.strictEqual(code, 0);
+          cb();
+        });
       }
     }
   },