Support Node.js 13.x

.travis.yml

@@ -9,6 +9,7 @@ node_js:
   - "8"
   - "10"
   - "12"
+  - "13"
 sudo: required
 services: docker
 before_install:

src/admin.cc

@@ -90,8 +90,8 @@ void AdminClient::Init(v8::Local<v8::Object> exports) {
 
   constructor.Reset(
     (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
-  exports->Set(Nan::New("AdminClient").ToLocalChecked(),
-    (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
+  Nan::Set(exports, Nan::New("AdminClient").ToLocalChecked(),
+    tpl->GetFunction(Nan::GetCurrentContext()).ToLocalChecked());
 }
 
 void AdminClient::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
@@ -547,9 +547,6 @@ NAN_METHOD(AdminClient::NodeCreatePartitions) {
   Nan::Callback *callback = new Nan::Callback(cb);
   AdminClient* client = ObjectWrap::Unwrap<AdminClient>(info.This());
 
-  // Get the timeout
-  int timeout = Nan::To<int32_t>(info[2]).FromJust();
-
   // Get the total number of desired partitions
   int partition_total_count = Nan::To<int32_t>(info[1]).FromJust();
 

src/binding.cc

@@ -64,24 +64,30 @@ void ConstantsInit(v8::Local<v8::Object> exports) {
   NODE_DEFINE_CONSTANT(topicConstants, RdKafka::Topic::OFFSET_STORED);
   NODE_DEFINE_CONSTANT(topicConstants, RdKafka::Topic::OFFSET_INVALID);
 
-  exports->Set(Nan::New("topic").ToLocalChecked(), topicConstants);
+  Nan::Set(exports, Nan::New("topic").ToLocalChecked(), topicConstants);
 
-  exports->Set(Nan::New("err2str").ToLocalChecked(),
+  Nan::Set(exports, Nan::New("err2str").ToLocalChecked(),
     Nan::GetFunction(Nan::New<v8::FunctionTemplate>(NodeRdKafkaErr2Str)).ToLocalChecked());  // NOLINT
 
-  exports->Set(Nan::New("features").ToLocalChecked(),
+  Nan::Set(exports, Nan::New("features").ToLocalChecked(),
     Nan::GetFunction(Nan::New<v8::FunctionTemplate>(NodeRdKafkaBuildInFeatures)).ToLocalChecked());  // NOLINT
 }
 
-void Init(v8::Local<v8::Object> exports, v8::Local<v8::Object> module) {
+void Init(v8::Local<v8::Object> exports, v8::Local<v8::Value> m_, void* v_) {
+#ifdef NODE_MAJOR_VERSION <= 8
   AtExit(RdKafkaCleanup);
+#else
+  v8::Local<v8::Context> context = Nan::GetCurrentContext();
+  node::Environment* env = node::GetCurrentEnvironment(context);
+  AtExit(env, RdKafkaCleanup, NULL);
+#endif
   KafkaConsumer::Init(exports);
   Producer::Init(exports);
   AdminClient::Init(exports);
   Topic::Init(exports);
   ConstantsInit(exports);
 
-  exports->Set(Nan::New("librdkafkaVersion").ToLocalChecked(),
+  Nan::Set(exports, Nan::New("librdkafkaVersion").ToLocalChecked(),
       Nan::New(RdKafka::version_str().c_str()).ToLocalChecked());
 }
 

src/callbacks.cc

@@ -42,7 +42,7 @@ v8::Local<v8::Array> TopicPartitionListToV8Array(
         Nan::New<v8::Number>(tp.offset));
     }
 
-    tp_array->Set(i, tp_obj);
+    Nan::Set(tp_array, i, tp_obj);
   }
 
   return tp_array;

src/common.cc

@@ -126,7 +126,11 @@ std::vector<std::string> v8ArrayToStringVector(v8::Local<v8::Array> parameter) {
 
   if (parameter->Length() >= 1) {
     for (unsigned int i = 0; i < parameter->Length(); i++) {
-      Nan::MaybeLocal<v8::String> p = Nan::To<v8::String>(parameter->Get(i));
+      v8::Local<v8::Value> v;
+      if (!Nan::Get(parameter, i).ToLocal(&v)) {
+        continue;
+      }
+      Nan::MaybeLocal<v8::String> p = Nan::To<v8::String>(v);
       if (p.IsEmpty()) {
         continue;
       }
@@ -146,7 +150,10 @@ std::vector<std::string> ToStringVector(v8::Local<v8::Array> parameter) {
 
   if (parameter->Length() >= 1) {
     for (unsigned int i = 0; i < parameter->Length(); i++) {
-      v8::Local<v8::Value> element = parameter->Get(i);
+      v8::Local<v8::Value> element;
+      if (!Nan::Get(parameter, i).ToLocal(&element)) {
+        continue;
+      }
 
       if (!element->IsRegExp()) {
         Nan::MaybeLocal<v8::String> p = Nan::To<v8::String>(element);
@@ -179,7 +186,7 @@ v8::Local<v8::Array> ToV8Array(std::vector<std::string> parameter) {
   for (size_t i = 0; i < parameter.size(); i++) {
     std::string topic = parameter[i];
 
-    newItem->Set(i, Nan::New<v8::String>(topic).ToLocalChecked());
+    Nan::Set(newItem, i, Nan::New<v8::String>(topic).ToLocalChecked());
   }
 
   return newItem;
@@ -203,7 +210,7 @@ v8::Local<v8::Array> ToV8Array(
       topic_partition_list[topic_partition_i];
 
     if (topic_partition->err() != RdKafka::ErrorCode::ERR_NO_ERROR) {
-      array->Set(topic_partition_i,
+      Nan::Set(array, topic_partition_i,
         Nan::Error(Nan::New(RdKafka::err2str(topic_partition->err()))
         .ToLocalChecked()));
     } else {
@@ -220,7 +227,7 @@ v8::Local<v8::Array> ToV8Array(
         Nan::New<v8::String>(topic_partition->topic().c_str())
         .ToLocalChecked());
 
-      array->Set(topic_partition_i, obj);
+      Nan::Set(array, topic_partition_i, obj);
     }
   }
 
@@ -243,8 +250,11 @@ std::vector<RdKafka::TopicPartition*> FromV8Array(
 
   for (size_t topic_partition_i = 0;
     topic_partition_i < topic_partition_list->Length(); topic_partition_i++) {
-    v8::Local<v8::Value> topic_partition_value =
-      topic_partition_list->Get(topic_partition_i);
+    v8::Local<v8::Value> topic_partition_value;
+    if (!Nan::Get(topic_partition_list, topic_partition_i)
+        .ToLocal(&topic_partition_value)) {
+      continue;
+    }
 
     if (topic_partition_value->IsObject()) {
       array.push_back(FromV8Object(
@@ -308,7 +318,7 @@ v8::Local<v8::Object> ToV8Object(RdKafka::Metadata* metadata) {
     Nan::Set(current_broker, Nan::New("port").ToLocalChecked(),
       Nan::New<v8::Number>(x->port()));
 
-    broker_data->Set(broker_i, current_broker);
+    Nan::Set(broker_data, broker_i, current_broker);
   }
 
   unsigned int topic_i = 0;
@@ -357,27 +367,27 @@ v8::Local<v8::Object> ToV8Object(RdKafka::Metadata* metadata) {
       v8::Local<v8::Array> current_replicas = Nan::New<v8::Array>();
 
       for (r_it = replicas->begin(); r_it != replicas->end(); ++r_it, r_i++) {
-        current_replicas->Set(r_i, Nan::New<v8::Int32>(*r_it));
+        Nan::Set(current_replicas, r_i, Nan::New<v8::Int32>(*r_it));
       }
 
       v8::Local<v8::Array> current_isrs = Nan::New<v8::Array>();
 
       for (i_it = isrs->begin(); i_it != isrs->end(); ++i_it, i_i++) {
-        current_isrs->Set(i_i, Nan::New<v8::Int32>(*i_it));
+        Nan::Set(current_isrs, i_i, Nan::New<v8::Int32>(*i_it));
       }
 
       Nan::Set(current_partition, Nan::New("replicas").ToLocalChecked(),
         current_replicas);
       Nan::Set(current_partition, Nan::New("isrs").ToLocalChecked(),
         current_isrs);
 
-      current_topic_partitions->Set(partition_i, current_partition);
+      Nan::Set(current_topic_partitions, partition_i, current_partition);
     }  // iterate over partitions
 
     Nan::Set(current_topic, Nan::New("partitions").ToLocalChecked(),
       current_topic_partitions);
 
-    topic_data->Set(topic_i, current_topic);
+    Nan::Set(topic_data, topic_i, current_topic);
   }  // End iterating over topics
 
   Nan::Set(obj, Nan::New("orig_broker_id").ToLocalChecked(),
@@ -431,7 +441,7 @@ v8::Local<v8::Object> ToV8Object(RdKafka::Message *message,
         Nan::Set(v8header, Nan::New<v8::String>(it->key()).ToLocalChecked(),
           Nan::Encode(it->value_string(),
             it->value_size(), Nan::Encoding::BUFFER));
-        v8headers->Set(index, v8header);
+        Nan::Set(v8headers, index, v8header);
         index++;
       }
       Nan::Set(pack,
@@ -519,7 +529,8 @@ rd_kafka_NewTopic_t* FromV8TopicObject(
     if (!config_keys.IsEmpty()) {
       v8::Local<v8::Array> field_array = config_keys.ToLocalChecked();
       for (size_t i = 0; i < field_array->Length(); i++) {
-        v8::Local<v8::String> config_key = field_array->Get(i).As<v8::String>();
+        v8::Local<v8::String> config_key = Nan::Get(field_array, i)
+          .ToLocalChecked().As<v8::String>();
         v8::Local<v8::Value> config_value = Nan::Get(config, config_key)
           .ToLocalChecked();
 
@@ -534,7 +545,7 @@ rd_kafka_NewTopic_t* FromV8TopicObject(
           err = rd_kafka_NewTopic_set_config(
             new_topic, pKeyString.c_str(), pValString.c_str());
 
-          if (err != RdKafka::ERR_NO_ERROR) {
+          if (err != RD_KAFKA_RESP_ERR_NO_ERROR) {
             errstr = rd_kafka_err2str(err);
             rd_kafka_NewTopic_destroy(new_topic);
             return NULL;

src/config.cc

@@ -35,6 +35,7 @@ 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));
 
   v8::MaybeLocal<v8::Array> _property_names = object->GetOwnPropertyNames(
@@ -45,8 +46,8 @@ Conf * Conf::create(RdKafka::Conf::ConfType type, v8::Local<v8::Object> object,
     std::string string_value;
     std::string string_key;
 
-    v8::Local<v8::Value> key = property_names->Get(i);
-    v8::Local<v8::Value> value = object->Get(key);
+    v8::Local<v8::Value> key = Nan::Get(property_names, i).ToLocalChecked();
+    v8::Local<v8::Value> value = Nan::Get(object, key).ToLocalChecked();
 
     if (key->IsString()) {
       Nan::Utf8String utf8_key(key);
@@ -59,13 +60,13 @@ Conf * Conf::create(RdKafka::Conf::ConfType type, v8::Local<v8::Object> object,
 #if NODE_MAJOR_VERSION > 6
       if (value->IsInt32()) {
         string_value = std::to_string(
-          value->Int32Value(Nan::GetCurrentContext()).ToChecked());
+          value->Int32Value(context).ToChecked());
       } else if (value->IsUint32()) {
         string_value = std::to_string(
-          value->Uint32Value(Nan::GetCurrentContext()).ToChecked());
+          value->Uint32Value(context).ToChecked());
       } else if (value->IsBoolean()) {
-        string_value = value->BooleanValue(
-          Nan::GetCurrentContext()).ToChecked() ? "true" : "false";
+        const bool v = Nan::To<bool>(value).ToChecked();
+        string_value = v ? "true" : "false";
       } else {
         Nan::Utf8String utf8_value(value.As<v8::String>());
         string_value = std::string(*utf8_value);

src/errors.cc

@@ -19,9 +19,9 @@ v8::Local<v8::Object> RdKafkaError(const RdKafka::ErrorCode &err, std::string er
 
   v8::Local<v8::Object> ret = Nan::New<v8::Object>();
 
-  ret->Set(Nan::New("message").ToLocalChecked(),
+  Nan::Set(ret, Nan::New("message").ToLocalChecked(),
     Nan::New<v8::String>(errstr).ToLocalChecked());
-  ret->Set(Nan::New("code").ToLocalChecked(),
+  Nan::Set(ret, Nan::New("code").ToLocalChecked(),
     Nan::New<v8::Number>(code));
 
   return ret;

src/kafka-consumer.cc

@@ -535,7 +535,7 @@ void KafkaConsumer::Init(v8::Local<v8::Object> exports) {
 
   constructor.Reset((tpl->GetFunction(Nan::GetCurrentContext()))
     .ToLocalChecked());
-  exports->Set(Nan::New("KafkaConsumer").ToLocalChecked(),
+  Nan::Set(exports, Nan::New("KafkaConsumer").ToLocalChecked(),
     (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
 }
 
@@ -712,8 +712,10 @@ NAN_METHOD(KafkaConsumer::NodeAssign) {
   std::vector<RdKafka::TopicPartition*> topic_partitions;
 
   for (unsigned int i = 0; i < partitions->Length(); ++i) {
-    v8::Local<v8::Value> partition_obj_value = partitions->Get(i);
-    if (!partition_obj_value->IsObject()) {
+    v8::Local<v8::Value> partition_obj_value;
+    if (!(
+          Nan::Get(partitions, i).ToLocal(&partition_obj_value) &&
+          partition_obj_value->IsObject())) {
       Nan::ThrowError("Must pass topic-partition objects");
     }
 

src/producer.cc

@@ -90,8 +90,8 @@ void Producer::Init(v8::Local<v8::Object> exports) {
   constructor.Reset((tpl->GetFunction(Nan::GetCurrentContext()))
     .ToLocalChecked());
 
-  exports->Set(Nan::New("Producer").ToLocalChecked(),
-    (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
+  Nan::Set(exports, Nan::New("Producer").ToLocalChecked(),
+    tpl->GetFunction(Nan::GetCurrentContext()).ToLocalChecked());
 }
 
 void Producer::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
@@ -450,17 +450,18 @@ NAN_METHOD(Producer::NodeProduce) {
 
     if (v8Headers->Length() >= 1) {
       for (unsigned int i = 0; i < v8Headers->Length(); i++) {
-        v8::Local<v8::Object> header = v8Headers->Get(i)->ToObject(
-          Nan::GetCurrentContext()).ToLocalChecked();
+        v8::Local<v8::Object> header = Nan::Get(v8Headers, i).ToLocalChecked()
+          ->ToObject(Nan::GetCurrentContext()).ToLocalChecked();
         if (header.IsEmpty()) {
           continue;
         }
 
         v8::Local<v8::Array> props = header->GetOwnPropertyNames(
           Nan::GetCurrentContext()).ToLocalChecked();
-        Nan::MaybeLocal<v8::String> v8Key = Nan::To<v8::String>(props->Get(0));
-        Nan::MaybeLocal<v8::String> v8Value =
-          Nan::To<v8::String>(header->Get(v8Key.ToLocalChecked()));
+        Nan::MaybeLocal<v8::String> v8Key = Nan::To<v8::String>(
+            Nan::Get(props, 0).ToLocalChecked());
+        Nan::MaybeLocal<v8::String> v8Value = Nan::To<v8::String>(
+            Nan::Get(header, v8Key.ToLocalChecked()).ToLocalChecked());
 
         Nan::Utf8String uKey(v8Key.ToLocalChecked());
         std::string key(*uKey);

src/topic.cc

@@ -89,8 +89,8 @@ void Topic::Init(v8::Local<v8::Object> exports) {
   constructor.Reset((tpl->GetFunction(Nan::GetCurrentContext()))
     .ToLocalChecked());
 
-  exports->Set(Nan::New("Topic").ToLocalChecked(),
-    (tpl->GetFunction(Nan::GetCurrentContext())).ToLocalChecked());
+  Nan::Set(exports, Nan::New("Topic").ToLocalChecked(),
+    tpl->GetFunction(Nan::GetCurrentContext()).ToLocalChecked());
 }
 
 void Topic::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {

src/workers.cc

@@ -569,7 +569,7 @@ void KafkaConsumerConsumeNum::HandleOKCallback() {
         it != m_messages.end(); ++it) {
       i++;
       RdKafka::Message* message = *it;
-      returnArray->Set(i, Conversion::Message::ToV8Object(message));
+      Nan::Set(returnArray, i, Conversion::Message::ToV8Object(message));
 
       delete message;
     }