Refactor config-based dispatchers to be controlled by config

Author: webmakersteve

lib/kafka-consumer.js

@@ -42,8 +42,30 @@ function KafkaConsumer(conf, topicConf) {
 
   var onRebalance = conf.rebalance_cb;
 
-  // Delete it because it messes other things up
-  delete conf.rebalance_cb;
+  var self = this;
+
+  // If rebalance is undefined we don't want any part of this
+  if (onRebalance && typeof onRebalance === 'boolean') {
+    conf.rebalance_cb = function(e) {
+      // That's it
+      if (e.code === 500 /*CODES.REBALANCE.PARTITION_ASSIGNMENT*/) {
+        self.assign(e.assignment);
+      } else if (e.code === 501 /*CODES.REBALANCE.PARTITION_UNASSIGNMENT*/) {
+        self.unassign(e.assignment);
+      }
+    };
+  } else if (onRebalance && typeof onRebalance === 'function') {
+    /*
+     * Once this is opted in to, that's it. It's going to manually rebalance
+     * forever. There is no way to unset config values in librdkafka, just
+     * a way to override them.
+     */
+
+     conf.rebalance_cb = function(e) {
+       self.emit('rebalance', e);
+       onRebalance.call(self, e);
+     };
+  }
 
 
   /**
@@ -63,42 +85,6 @@ function KafkaConsumer(conf, topicConf) {
 
   Client.call(this, conf, Kafka.KafkaConsumer, topicConf);
 
-  var self = this;
-
-  if (onRebalance !== false) {
-    // If rebalance is specifically false we don't want any part of this
-
-    if (typeof onRebalance !== 'function') {
-      onRebalance = function(e) {
-        // That's it
-        self.assign(e.assignments);
-      };
-    }
-
-  }
-
-  /*
-   * Once this is opted in to, that's it. It's going to manually rebalance
-   * forever. There is no way to unset config values in librdkafka, just
-   * a way to override them.
-   */
-
-  if (onRebalance) {
-    /**
-     * Rebalance event. Called when the KafkaConsumer is rebalancing.
-     *
-     * @event KafkaConsumer#rebalance
-     * @type {object}
-     * @property {number} code - whether the rebalance was an assignment or
-     * an unassignment
-     */
-    this._client.onRebalance(function(e) {
-      self.emit('rebalance', e);
-      onRebalance.call(self, e);
-    });
-
-  }
-
   this.globalConfig = conf;
   this.topicConfig = topicConf;
 }

src/callbacks.cc

@@ -386,7 +386,13 @@ void Delivery::dr_cb(RdKafka::Message &message) {
 
 // Rebalance CB
 
-/*
+RebalanceDispatcher::RebalanceDispatcher() {}
+RebalanceDispatcher::~RebalanceDispatcher() {}
+
+void RebalanceDispatcher::Add(const rebalance_event_t &e) {
+  scoped_mutex_lock lock(async_lock);
+  events.push_back(e);
+}
 
 void RebalanceDispatcher::Flush() {
   Nan::HandleScope scope;
@@ -447,16 +453,18 @@ void RebalanceDispatcher::Flush() {
     Dispatch(argc, argv);
   }
 }
-*/
-Rebalance::Rebalance(Nan::Callback &cb) {}
+
+Rebalance::Rebalance(v8::Local<v8::Function> &cb) {
+  dispatcher.AddCallback(cb);
+}
 Rebalance::~Rebalance() {}
 
 void Rebalance::rebalance_cb(RdKafka::KafkaConsumer *consumer,
   RdKafka::ErrorCode err,
   std::vector<RdKafka::TopicPartition*> &partitions) {
 
-  // dispatcher.Add(rebalance_event_t(err, partitions));
-  // dispatcher.Execute();
+  dispatcher.Add(rebalance_event_t(err, partitions));
+  dispatcher.Execute();
 
 }
 

src/callbacks.h

@@ -174,14 +174,25 @@ struct rebalance_event_t {
   }
 };
 
+class RebalanceDispatcher : public Dispatcher {
+ public:
+  RebalanceDispatcher();
+  ~RebalanceDispatcher();
+  void Add(const rebalance_event_t &);
+  void Flush();
+ protected:
+  std::vector<rebalance_event_t> events;
+};
 
 class Rebalance : public RdKafka::RebalanceCb {
  public:
-  explicit Rebalance(Nan::Callback &);
+  explicit Rebalance(v8::Local<v8::Function>&);
   ~Rebalance();
 
   void rebalance_cb(RdKafka::KafkaConsumer *, RdKafka::ErrorCode,
     std::vector<RdKafka::TopicPartition*> &);
+
+  RebalanceDispatcher dispatcher;
  private:
   v8::Persistent<v8::Function> m_cb;
 };

src/config.cc

@@ -62,11 +62,10 @@ Conf * Conf::create(RdKafka::Conf::ConfType type, v8::Local<v8::Object> object,
           return NULL;
       }
     } else {
-      Log("Value is a function");
       if (string_key.compare("rebalance_cb") == 0) {
-        Nan::Callback cb(value.As<v8::Function>());
-        NodeKafka::Callbacks::Rebalance rebalance_cb(cb);
-        rdconf->set(string_key, &rebalance_cb, errstr);
+        v8::Local<v8::Function> cb = value.As<v8::Function>();
+        rdconf->m_rebalance_cb = new NodeKafka::Callbacks::Rebalance(cb);
+        rdconf->set(string_key, rdconf->m_rebalance_cb, errstr);
       }
     }
   }
@@ -75,6 +74,22 @@ Conf * Conf::create(RdKafka::Conf::ConfType type, v8::Local<v8::Object> object,
 
 }
 
-Conf::~Conf() {}
+void Conf::listen() {
+  if (m_rebalance_cb) {
+    m_rebalance_cb->dispatcher.Activate();
+  }
+}
+
+void Conf::stop() {
+  if (m_rebalance_cb) {
+    m_rebalance_cb->dispatcher.Deactivate();
+  }
+}
+
+Conf::~Conf() {
+  if (m_rebalance_cb) {
+    delete m_rebalance_cb;
+  }
+}
 
 }  // namespace NodeKafka

src/config.h

@@ -28,9 +28,11 @@ class Conf : public RdKafka::Conf {
   static Conf* create(RdKafka::Conf::ConfType, v8::Local<v8::Object>, std::string &);  // NOLINT
 
   static void DumpConfig(std::list<std::string> *);
+
+  void listen();
+  void stop();
  protected:
-  bool m_has_rebalance_cb;
-  bool m_has_partitioner_cb;
+  NodeKafka::Callbacks::Rebalance * m_rebalance_cb = NULL;
 };
 
 }  // namespace NodeKafka

src/connection.cc

@@ -36,7 +36,7 @@ namespace NodeKafka {
  * @sa NodeKafka::Client
  */
 
-Connection::Connection(RdKafka::Conf* gconfig, RdKafka::Conf* tconfig):
+Connection::Connection(Conf* gconfig, Conf* tconfig):
   m_event_cb(),
   m_gconfig(gconfig),
   m_tconfig(tconfig) {

src/connection.h

@@ -63,7 +63,7 @@ class Connection : public Nan::ObjectWrap {
   virtual void DeactivateDispatchers() = 0;
 
  protected:
-  Connection(RdKafka::Conf*, RdKafka::Conf*);
+  Connection(Conf*, Conf*);
   ~Connection();
 
   static Nan::Persistent<v8::Function> constructor;
@@ -72,8 +72,8 @@ class Connection : public Nan::ObjectWrap {
   bool m_has_been_disconnected;
   bool m_is_closing;
 
-  RdKafka::Conf* m_gconfig;
-  RdKafka::Conf* m_tconfig;
+  Conf* m_gconfig;
+  Conf* m_tconfig;
   std::string m_errstr;
 
   uv_mutex_t m_connection_lock;

src/consumer.cc

@@ -37,11 +37,8 @@ consumer_commit_t::consumer_commit_t() {
  * @sa NodeKafka::Client
  */
 
-Consumer::Consumer(RdKafka::Conf* gconfig, RdKafka::Conf* tconfig):
+Consumer::Consumer(Conf* gconfig, Conf* tconfig):
   Connection(gconfig, tconfig) {
-    m_is_subscribed = false;
-    m_is_manually_rebalancing = false;
-
     std::string errstr;
 
     m_gconfig->set("default_topic_conf", m_tconfig, errstr);
@@ -73,6 +70,13 @@ Baton Consumer::Connect() {
 }
 
 void Consumer::ActivateDispatchers() {
+  // Listen to global config
+  m_gconfig->listen();
+
+  // Listen to non global config
+  // tconfig->listen();
+
+  // This should be refactored to config based management
   m_event_cb.dispatcher.Activate();
 }
 
@@ -100,6 +104,10 @@ Baton Consumer::Disconnect() {
 }
 
 void Consumer::DeactivateDispatchers() {
+  // Stop listening to the config dispatchers
+  m_gconfig->stop();
+
+  // Also this one
   m_event_cb.dispatcher.Deactivate();
 }
 

src/consumer.h

@@ -79,16 +79,15 @@ class Consumer : public Connection {
   static Nan::Persistent<v8::Function> constructor;
   static void New(const Nan::FunctionCallbackInfo<v8::Value>& info);
 
-  Consumer(RdKafka::Conf *, RdKafka::Conf *);
+  Consumer(Conf *, Conf *);
   ~Consumer();
 
  private:
   static void part_list_print(const std::vector<RdKafka::TopicPartition*>&);
 
   std::vector<RdKafka::TopicPartition*> m_partitions;
   int m_partition_cnt;
-  bool m_is_subscribed;
-  bool m_is_manually_rebalancing;
+  bool m_is_subscribed = false;
 
   // Node methods
   static NAN_METHOD(NodeConnect);

src/producer.cc

@@ -28,7 +28,7 @@ namespace NodeKafka {
  * @sa NodeKafka::Connection
  */
 
-Producer::Producer(RdKafka::Conf* gconfig, RdKafka::Conf* tconfig):
+Producer::Producer(Conf* gconfig, Conf* tconfig):
   Connection(gconfig, tconfig),
   m_dr_cb(),
   m_partitioner_cb() {
@@ -109,14 +109,14 @@ void Producer::New(const Nan::FunctionCallbackInfo<v8::Value>& info) {
 
   std::string errstr;
 
-  RdKafka::Conf* gconfig =
+  Conf* gconfig =
     Conf::create(RdKafka::Conf::CONF_GLOBAL, info[0]->ToObject(), errstr);
 
   if (!gconfig) {
     return Nan::ThrowError(errstr.c_str());
   }
 
-  RdKafka::Conf* tconfig =
+  Conf* tconfig =
     Conf::create(RdKafka::Conf::CONF_TOPIC, info[1]->ToObject(), errstr);
 
   if (!tconfig) {