Allow topic config properties to be set with the global config [c++]

Author: milindl

index.d.ts

@@ -203,7 +203,7 @@ export abstract class Client<Events extends string> extends EventEmitter {
 }
 
 export class KafkaConsumer extends Client<KafkaConsumerEvents> {
-    constructor(conf: ConsumerGlobalConfig, topicConf: ConsumerTopicConfig);
+    constructor(conf: ConsumerGlobalConfig | ConsumerTopicConfig, topicConf?: ConsumerTopicConfig);
 
     assign(assignments: Assignment[]): this;
 
@@ -256,7 +256,7 @@ export class KafkaConsumer extends Client<KafkaConsumerEvents> {
 }
 
 export class Producer extends Client<KafkaProducerEvents> {
-    constructor(conf: ProducerGlobalConfig, topicConf?: ProducerTopicConfig);
+    constructor(conf: ProducerGlobalConfig | ProducerTopicConfig, topicConf?: ProducerTopicConfig);
 
     flush(timeout?: NumberNullUndefined, cb?: (err: LibrdKafkaError) => void): this;
 

lib/client.js

@@ -45,7 +45,6 @@ function Client(globalConf, SubClientType, topicConf) {
   // the producer and consumer main wrappers
 
   var no_event_cb = globalConf.event_cb === false;
-  topicConf = topicConf || {};
 
   // delete this because librdkafka will complain since this particular
   // key is a real conf value

src/kafka-consumer.cc

@@ -31,7 +31,8 @@ KafkaConsumer::KafkaConsumer(Conf* gconfig, Conf* tconfig):
   Connection(gconfig, tconfig) {
     std::string errstr;
 
-    m_gconfig->set("default_topic_conf", m_tconfig, errstr);
+    if (m_tconfig)
+      m_gconfig->set("default_topic_conf", m_tconfig, errstr);
 
     m_consume_loop = nullptr;
   }
@@ -553,10 +554,6 @@ void KafkaConsumer::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
     return Nan::ThrowError("Global configuration data must be specified");
   }
 
-  if (!info[1]->IsObject()) {
-    return Nan::ThrowError("Topic configuration must be specified");
-  }
-
   std::string errstr;
 
   Conf* gconfig =
@@ -567,13 +564,16 @@ void KafkaConsumer::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
     return Nan::ThrowError(errstr.c_str());
   }
 
-  Conf* tconfig =
-    Conf::create(RdKafka::Conf::CONF_TOPIC,
+  // If tconfig isn't set, then just let us pick properties from gconf.
+  Conf* tconfig = nullptr;
+  if (info[1]->IsObject()) {
+    tconfig = Conf::create(RdKafka::Conf::CONF_TOPIC,
       (info[1]->ToObject(Nan::GetCurrentContext())).ToLocalChecked(), errstr);
 
-  if (!tconfig) {
-    delete gconfig;
-    return Nan::ThrowError(errstr.c_str());
+    if (!tconfig) {
+      delete gconfig;
+      return Nan::ThrowError(errstr.c_str());
+    }
   }
 
   KafkaConsumer* consumer = new KafkaConsumer(gconfig, tconfig);

src/producer.cc

@@ -35,7 +35,9 @@ Producer::Producer(Conf* gconfig, Conf* tconfig):
   m_partitioner_cb() {
     std::string errstr;
 
-    m_gconfig->set("default_topic_conf", m_tconfig, errstr);
+    if (m_tconfig)
+      m_gconfig->set("default_topic_conf", m_tconfig, errstr);
+
     m_gconfig->set("dr_cb", &m_dr_cb, errstr);
   }
 
@@ -110,10 +112,6 @@ void Producer::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
     return Nan::ThrowError("Global configuration data must be specified");
   }
 
-  if (!info[1]->IsObject()) {
-    return Nan::ThrowError("Topic configuration must be specified");
-  }
-
   std::string errstr;
 
   Conf* gconfig =
@@ -124,14 +122,17 @@ void Producer::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
     return Nan::ThrowError(errstr.c_str());
   }
 
-  Conf* tconfig =
-    Conf::create(RdKafka::Conf::CONF_TOPIC,
-      (info[1]->ToObject(Nan::GetCurrentContext())).ToLocalChecked(), errstr);
+  // If tconfig isn't set, then just let us pick properties from gconf.
+  Conf* tconfig = nullptr;
+  if (info[1]->IsObject()) {
+    tconfig = Conf::create(RdKafka::Conf::CONF_TOPIC,
+                 (info[1]->ToObject(Nan::GetCurrentContext())).ToLocalChecked(), errstr);
 
-  if (!tconfig) {
-    // No longer need this since we aren't instantiating anything
-    delete gconfig;
-    return Nan::ThrowError(errstr.c_str());
+    if (!tconfig) {
+      // No longer need this since we aren't instantiating anything
+      delete gconfig;
+      return Nan::ThrowError(errstr.c_str());
+    }
   }
 
   Producer* producer = new Producer(gconfig, tconfig);