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

@@ -12,7 +12,6 @@
 
 #include <nan.h>
 
-#include <iostream>
 #include <string>
 #include <vector>
 

src/common.h

@@ -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) {