Support for several event types in the same topic

.gitignore

@@ -3,3 +3,6 @@ npm-debug.log
 dump.rdb
 wiki
 temp
+report.html
+coverage
+.nyc_output

README.md

@@ -2,7 +2,7 @@
 
 > Node.js bindings for librdkafka with Avro schema serialization.
 
-[![CircleCI](https://circleci.com/gh/waldophotos/kafka-avro/tree/master.svg?style=svg)](https://circleci.com/gh/waldophotos/kafka-avro/tree/master)
+[![CircleCI](https://circleci.com/gh/pleszczy/kafka-avro/tree/master.svg?style=svg)](https://circleci.com/gh/pleszczy/kafka-avro/tree/master)
 
 The kafka-avro library is a wrapper that combines the [node-rdkafka][node-rdkafka] and [avsc][avsc] libraries to allow for Production and Consumption of messages on kafka validated and serialized by Avro.
 
@@ -31,17 +31,17 @@ You are highly encouraged to read the ["node-rdkafka" documentation](https://bli
 The `Kafka.CODES` enumeration of constant values provided by the "node-rdkafka" library is also available as a static var at:
 
 ```js
-var KafkaAvro = require('kafka-avro');
+const KafkaAvro = require('kafka-avro');
 
 console.log(KafkaAvro.CODES);
 ```
 
 ### Initialize kafka-avro
 
 ```js
-var KafkaAvro = require('kafka-avro');
+const KafkaAvro = require('kafka-avro');
 
-var kafkaAvro = new KafkaAvro({
+const kafkaAvro = new KafkaAvro({
     kafkaBroker: 'localhost:9092',
     schemaRegistry: 'http://localhost:8081',
 });
@@ -78,17 +78,17 @@ kafkaAvro.getProducer({
 })
     // "getProducer()" returns a Bluebird Promise.
     .then(function(producer) {
-        var topicName = 'test';
+        const topicName = 'test';
 
         producer.on('disconnected', function(arg) {
           console.log('producer disconnected. ' + JSON.stringify(arg));
         });
 
-        var value = {name:'John'};
-        var key = 'key';
+        const value = {name:'John'};
+        const key = 'key';
 
         // if partition is set to -1, librdkafka will use the default partitioner
-        var partition = -1;
+        const partition = -1;
         producer.produce(topicName, partition, value, key);
     })
 ```
@@ -133,7 +133,7 @@ kafkaAvro.getConsumer({
   })
   .then(function(consumer) {
     // Subscribe and consume.
-    var topicName = 'test';
+    const topicName = 'test';
     consumer.subscribe([topicName]);
     consumer.consume();
     consumer.on('data', function(rawData) {
@@ -200,6 +200,12 @@ kafka-avro intercepts all incoming messages and augments the object with two mor
 
 The KafkaAvro instance also provides the following methods:
 
+### Support for several event types in the same topic
+The Kafka Avro supports several events types in the same topic. The strategy chosen to implement this is [TopicRecordNameStrategy](https://github.com/confluentinc/schema-registry/blob/master/avro-serializer/src/main/java/io/confluent/kafka/serializers/subject/TopicRecordNameStrategy.java)
+which is default strategy on the confluent platform.
+
+You can read more about this here : https://www.confluent.io/blog/put-several-event-types-kafka-topic/
+
 ### Logging
 
 The Kafka Avro library logs messages using the [Bunyan logger](https://github.com/trentm/node-bunyan/). To enable logging you will have to define at least one of the needed ENV variables:
@@ -218,10 +224,10 @@ The Kafka Avro library logs messages using the [Bunyan logger](https://github.co
 **Returns** {Bunyan.Logger} [Bunyan logger](https://github.com/trentm/node-bunyan/) instance.
 
 ```js
-var KafkaAvro = require('kafka-avro');
-var fmt = require('bunyan-format');
+const KafkaAvro = require('kafka-avro');
+const fmt = require('bunyan-format');
 
-var kafkaLog  = KafkaAvro.getLogger();
+const kafkaLog  = KafkaAvro.getLogger();
 
 kafkaLog.addStream({
     type: 'stream',

circle.yml

@@ -1,16 +1,42 @@
-machine:
-  pre:
-   - curl -sSL https://s3.amazonaws.com/circle-downloads/install-circleci-docker.sh | bash -s -- 1.10.0
-   - pip install docker-compose
-  services:
-    - docker
-  node:
-    version: 8.2.0
-  hosts:
-    kafka: 127.0.0.1
-
-dependencies:
-  pre:
-    - docker-compose up -d
-    - sudo apt-get update;
-    - sudo apt-get -y install libsasl2-dev libssl-dev
+version: 2
+jobs:
+  build:
+    working_directory: ~/kafka-avro
+    machine:
+      image: ubuntu-1604:201903-01
+      docker_layer_caching: true
+    steps:
+      - checkout
+      - run:
+          name: Start kafka and schema-registry
+          command: docker-compose up -d
+      - run:
+          name: Install [email protected]
+          command: |
+            set +e
+            curl -o- https://raw.githubusercontent.com/creationix/nvm/v0.33.5/install.sh | bash
+            export NVM_DIR="/opt/circleci/.nvm"
+            [ -s "$NVM_DIR/nvm.sh" ] && \. "$NVM_DIR/nvm.sh"
+            nvm install v8.16
+            nvm alias default v8.16
+            # Each step uses the same `$BASH_ENV`, so need to modify it
+            echo 'export NVM_DIR="/opt/circleci/.nvm"' >> $BASH_ENV
+            echo "[ -s \"$NVM_DIR/nvm.sh\" ] && . \"$NVM_DIR/nvm.sh\"" >> $BASH_ENV
+      - restore_cache:
+          key: dependency-cache-{{ checksum "package.json" }}
+      - save_cache:
+          key: dependency-cache-{{ checksum "package.json" }}
+          paths:
+            - node_modules
+      - run:
+          name: Install projects dependencies
+          command: 'unset NVM_NODEJS_ORG_MIRROR NVM_IOJS_ORG_MIRROR && npm install'
+      - run:
+          name: Add hostnames for kafka
+          command: echo 127.0.0.1 kafka | sudo tee -a /etc/hosts && echo 127.0.0.1 schema-registry | sudo tee -a /etc/hosts && echo 127.0.0.1 zookeeper | sudo tee -a /etc/hosts
+      - run:
+          name: print /etc/hosts content
+          command: cat /etc/hosts
+      - run:
+          name: Test
+          command: npm run-script coverage

lib/kafka-avro.js

@@ -6,29 +6,27 @@
  * Copyright © Waldo, Inc.
  * Licensed under the MIT license.
  */
-var EventEmitter = require('events').EventEmitter;
-
-var Promise = require('bluebird');
-var cip = require('cip');
-var Kafka = require('node-rdkafka');
-
-var rootLog = require('./log.lib');
-var log = rootLog.getChild(__filename);
-
-var SchemaRegistry = require('./schema-registry');
+const EventEmitter = require('events').EventEmitter;
+const Promise = require('bluebird');
+const cip = require('cip');
+const Kafka = require('node-rdkafka');
+const rootLog = require('./log.lib');
+const log = rootLog.getChild(__filename);
+const SchemaRegistry = require('./schema-registry');
 
 //
 // Mixins
 //
-var Producer = require('./kafka-producer');
-var Consumer = require('./kafka-consumer');
+const Producer = require('./kafka-producer');
+const Consumer = require('./kafka-consumer');
 
-var CeventEmitter = cip.cast(EventEmitter);
+const CeventEmitter = cip.cast(EventEmitter);
 
-function noop() {}
+function noop() {
+}
 
 /**
- * @fileOverview bootstrap and master exporing module.
+ * @fileOverview bootstrap and master exporting module.
  */
 
 /**
@@ -37,11 +35,11 @@ function noop() {}
  * @param {Object} opts The options.
  * @constructor
  */
-var KafkaAvro = module.exports = CeventEmitter.extend(function(opts) {
+const KafkaAvro = module.exports = CeventEmitter.extend(function (opts) {
   /** @type {string} The SR url */
   this.kafkaBrokerUrl = opts.kafkaBroker;
 
-  var srOpts = {
+  const srOpts = {
     schemaRegistryUrl: opts.schemaRegistry,
     selectedTopics: opts.topics || null,
     fetchAllVersions: opts.fetchAllVersions || false,
@@ -50,14 +48,14 @@ var KafkaAvro = module.exports = CeventEmitter.extend(function(opts) {
     httpsAgent: opts.httpsAgent
   };
 
-  /** @type {kafka-avro.SchemaRegistry} Instanciated SR. */
+  /** @type {kafka-avro.SchemaRegistry} Instantiated SR. */
   this.sr = new SchemaRegistry(srOpts);
 
-  /** @type {Array.<node-rdkafka.Producer>} Instanciated producers. */
+  /** @type {Array.<node-rdkafka.Producer>} Instantiated producers. */
   this._producers = [];
-  /** @type {Array.<node-rdkafka.Consumer>} Instanciated consumers. */
+  /** @type {Array.<node-rdkafka.Consumer>} Instantiated consumers. */
   this._consumers = [];
-  /** @type {Array.<node-rdkafka.ConsumerStream>} Instanciated consumers. */
+  /** @type {Array.<node-rdkafka.ConsumerStream>} Instantiated consumers. */
   this._consumersStream = [];
 });
 
@@ -68,9 +66,6 @@ var KafkaAvro = module.exports = CeventEmitter.extend(function(opts) {
  */
 KafkaAvro.CODES = Kafka.CODES;
 
-//
-// Add Mixins
-//
 KafkaAvro.mixin(Producer);
 KafkaAvro.mixin(Consumer);
 
@@ -100,27 +95,29 @@ KafkaAvro.prototype.init = Promise.method(function () {
  * @return {Promise} A Promise.
  */
 KafkaAvro.prototype.dispose = Promise.method(function () {
+  const disconnectPromises = [];
 
-  var disconnectPromises = [];
-
-  log.info('dispose() :: Disposing kafka-avro instance. Total consumers:',
-    this._consumers.length, 'Total producers:', this._producers.length);
+  log.info('dispose() :: Disposing kafka-avro instance. Total consumers: noOfConsumers, Total producers: noOfProducers',
+    {
+      noOfConsumers: this._consumers.length,
+      noOfProducers: this._producers.length
+    });
 
-  this._consumers.forEach(function(consumer) {
-    var discon = Promise.promisify(consumer.disconnect.bind(consumer));
-    var disconProm = discon().catch(noop);
+  this._consumers.forEach(function (consumer) {
+    const discon = Promise.promisify(consumer.disconnect.bind(consumer));
+    const disconProm = discon().catch(noop);
 
     disconnectPromises.push(disconProm);
   });
-  this._producers.forEach(function(producer) {
-    var discon = Promise.promisify(producer.disconnect.bind(producer));
-    var disconProm = discon().catch(noop);
+  this._producers.forEach(function (producer) {
+    const discon = Promise.promisify(producer.disconnect.bind(producer));
+    const disconProm = discon().catch(noop);
     disconnectPromises.push(disconProm);
   });
 
   return Promise.all(disconnectPromises)
-    .then( () => { 
-      if(this.sr._refreshHandle) 
-        clearInterval(this.sr._refreshHandle); 
+    .then(() => {
+      if (this.sr._refreshHandle)
+        clearInterval(this.sr._refreshHandle);
     });
 });