Merge pull request #159 from svarkey/kafka_use_high_level_consumer

Kafka use high level consumer

lib/kafka_ascoltatore.js

@@ -157,6 +157,7 @@ KafkaAscoltatore.prototype._startConn = function(cb) {
    var noAckBatchOptions = that._opts.noAckBatchOptions;
    var groupId = that._opts.groupId || that._default_opts.groupId;
    var Client = this._opts.kafka.Client;
+   var useHighLevelConsumer = that._opts.useHighLevelConsumer || false;
    var HighLevelConsumer = this._opts.kafka.HighLevelConsumer;
    var Consumer = this._opts.kafka.Consumer;
    var subscribedTopics = that._subs_counter.keys();
@@ -173,9 +174,16 @@ KafkaAscoltatore.prototype._startConn = function(cb) {
           return false;
         }
         debug("initial subscriptions expanded to ",subscriptions);
-        that._consumer = new Consumer(that._consumerClient,subscriptions,{groupId: groupId,
-                fromOffset: true,    autoCommit: false, encoding: "buffer" });
-
+        if (useHighLevelConsumer) {
+           debug("Using high level consumer ");
+           that._consumer = new HighLevelConsumer(that._consumerClient, subscriptions, {
+               groupId: groupId,fromOffset: true, autoCommit: false, encoding: "buffer"
+           });
+        } else {
+           debug("Using simple consumer ");
+           that._consumer = new Consumer(that._consumerClient, subscriptions, {
+               groupId: groupId,fromOffset: true, autoCommit: false, encoding: "buffer"});
+        }
         that._consumer.on("message", function(message) {
             debug("received new message on topic ", message.topic);
             var value = message.value;

test/common.js

@@ -1,5 +1,4 @@
 "use strict";
-
 global.sinon = require("sinon");
 global.chai = require("chai");
 global.expect = require("chai").expect;
@@ -27,13 +26,14 @@ global.zeromqSettings = function(remote_ports) {
   };
 };
 
-global.kafkaSettings = function() {
+global.kafkaSettings = function(useHighLevelConsumer) {
   return {
     json: false,
     kafka: require("kafka-node"),
     connectionString: "localhost:2181",
     clientId: "test",
     groupId: "test",
+    useHighLevelConsumer: useHighLevelConsumer || false ,
     defaultEncoding: "utf8",
     encodings: {image: "buffer", hello_42: "utf-8"}
   };

test/kafka_ascoltatore_spec.js

@@ -1,6 +1,7 @@
 var fs = require("fs");
 var util = require("../lib/util");
 var steed = require('steed')();
+var kafka =  require("kafka-node");
 
 describeAscoltatore("kafka", function() {
 
@@ -47,5 +48,25 @@ describeAscoltatore("kafka", function() {
       that.instance.pub("hello", "€99");
     });
   });
+  it("Use high level consumer to sync two instances", function(done) {
+      var other = new ascoltatori.KafkaAscoltatore(kafkaSettings(true));
+      var HighLevelConsumer = kafka.HighLevelConsumer;
+      var that = this;
+      steed.series([
+          function(cb) {
+              other.on("ready", cb);
+          },
+          function(cb) {
+              other.subscribe("hello", function() {}, cb);
+          },
+          function(cb) {
+              that.instance.publish("hello", null, cb);
+          },
+          function() {
+              expect(other._consumer).to.be.an.instanceof(HighLevelConsumer);
+              done();
+          }
+      ]);
+  });
 
 });