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

@@ -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',
 });
@@ -66,6 +66,8 @@ When instantiating kafka-avro you may pass the following options:
 * `parseOptions` **Object** Schema parse options to pass to `avro.parse()`. `parseOptions.wrapUnions` is set to `true` by default.
 * `httpsAgent` **Object** initialized [https Agent class](https://nodejs.org/api/https.html#https_class_https_agent)
 * `shouldFailWhenSchemaIsMissing` **Boolean** Set to true if producing a message for which no AVRO schema can be found should throw an error
+* `keySubjectStrategy` **String** A SubjectNameStrategy for key. It is used by the Avro serializer to determine the subject name under which the event record schemas should be registered in the schema registry. The default is TopicNameStrategy. Allowed values are [TopicRecordNameStrategy, TopicNameStrategy, RecordNameStrategy]
+* `valueSubjectStrategy` **String** **String** A SubjectNameStrategy for value. It is used by the Avro serializer to determine the subject name under which the event record schemas should be registered in the schema registry. The default is TopicNameStrategy. Allowed values are [TopicRecordNameStrategy, TopicNameStrategy, RecordNameStrategy]
 
 ### Producer
 
@@ -79,17 +81,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);
     })
 ```
@@ -134,7 +136,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) {
@@ -201,6 +203,29 @@ 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
+Kafka Avro can support several events types in the same topic. This requires using TopicRecordNameStrategy strategy.
+
+```js
+const KafkaAvro = require('kafka-avro');
+
+const kafkaAvro = new KafkaAvro({
+    kafkaBroker: 'localhost:9092',
+    schemaRegistry: 'http://localhost:8081',
+    keySubjectStrategy: "TopicRecordNameStrategy",
+    valueSubjectStrategy: "TopicRecordNameStrategy",
+});
+
+// Query the Schema Registry for all topic-schema's
+// fetch them and evaluate them.
+kafkaAvro.init()
+    .then(function() {
+        console.log('Ready to use');
+    });
+```
+
+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:
@@ -219,10 +244,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,28 @@
  * 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');
+const { SubjectNameStrategy } = require('./subject-strategy');
 
 //
 // 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,30 +36,32 @@ 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,
     fetchRefreshRate: opts.fetchRefreshRate || 0,
     parseOptions: opts.parseOptions,
-    httpsAgent: opts.httpsAgent
+    httpsAgent: opts.httpsAgent,
+    keySubjectStrategy: new SubjectNameStrategy(opts.keySubjectStrategy),
+    valueSubjectStrategy: new SubjectNameStrategy(opts.valueSubjectStrategy),
   };
 
-  /** @type {kafka-avro.SchemaRegistry} Instanciated SR. */
+  /** @type {kafka-avro.SchemaRegistry} Instantiated SR. */
   this.sr = new SchemaRegistry(srOpts);
 
   /** @type {boolean} Whether the producer should fail when no schema was found. */
   this._shouldFailWhenSchemaIsMissing = opts.shouldFailWhenSchemaIsMissing === true;
 
   /** @type {Array.<node-rdkafka.Producer>} Instanciated 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 = [];
 });
 
@@ -71,9 +72,6 @@ var KafkaAvro = module.exports = CeventEmitter.extend(function(opts) {
  */
 KafkaAvro.CODES = Kafka.CODES;
 
-//
-// Add Mixins
-//
 KafkaAvro.mixin(Producer);
 KafkaAvro.mixin(Consumer);
 
@@ -103,21 +101,23 @@ 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);
   });