Defitions, Configuration, Disconnect, etc

Author: opiepj

.env.json

@@ -0,0 +1,10 @@
+{
+  "Topic1Name":"kafka-test-topic",
+  "Topic1Partition":"1",
+  "Topic1Replicas":"1",
+  "ZookeeperPort":"2181",
+  "KafkaProtocol":"kafka",
+  "KafkaPort":"9092",
+  "KafkaIP":"localhost",
+  "Throttle":"250"
+}

API.md

@@ -1,5 +1,35 @@
+<a name="top"></a>
+
 # API
 
+* [Configuration](#Configuration)
+* [Consumer](#Consumer)
+* [Producer](#Producer)
+
+
+<a name="Configuration"></a>
+
+## Configuration
+
+### All
+
+Configurations custom to this wrapper
+
+|   Field    | Description|   Type     | Default   |
+|------------|------------|------------|-----------|
+| throttle   | Throttle interval time (ms) | Number | 500|
+| topics     | Topics to subscribe to  | String[]  | \['kafka-test-topic'\] |
+| autoInterval | Allow auto intervals for polling (producer) and consuming (consumer). | boolean  | true |
+
+### Consumer
+
+Configurations custom to this wrapper's Consumer class.
+
+|   Field    | Description|   Type     | Default   |
+|------------|------------|------------|-----------|
+| consumeMax   | Number of messages to consume for each interval. | Number | 1 |
+
+[back to top](#top)
 
 <a name="Consumer"></a>
 

src/client.js

@@ -72,7 +72,6 @@ class KafkaClient {
     }
     return new Promise(
       (resolve, reject) => {
-        kafkaClient.disconnect();
 
         kafkaClient.prependListener('event.error', (err) => {
           console.error('Disconnect Operation (Error)', `${new Date()}: Error:`, err);
@@ -84,6 +83,14 @@ class KafkaClient {
           console.log('Disconnect Operation', `${new Date()}: Client Disconnected: ${JSON.stringify(arg)}`);
           return resolve();
         });
+
+        kafkaClient.disconnect();
+
+        setTimeout(() => {
+          const err = new Error(`Disconnect Operation (Error) ${new Date()}: Failed to Disconnect`);
+          console.log(err.message);
+          return reject(err);
+        }, 20000); // Wait 20 seconds
       })
       .then(() => {
         KAFKA_EVENTS.forEach((event) => {
@@ -113,6 +120,10 @@ class KafkaClient {
       console.error(`Error Log: ${new Date()}:`, err);
       this._errorDispatcher.next(err);
     });
+
+    kafkaClient.on('disconnected', (arg) => {
+
+    })
   }
 
   /**

src/consumer.js

@@ -110,7 +110,7 @@ class Consumer extends KafkaClient {
     this._consumeLoop = setInterval(() => {
       // start consuming messages
       this.kafkaConsumer.consume(this._config.consumeMax);
-    }, this._config.throttle);
+    }, this._config.e);
   }
 
 }

src/default-config.js

@@ -1,5 +1,17 @@
-const ENV = require('dotenv').config().parsed; // Environment from docker .ENV
+// const ENV = require('dotenv').config().parsed; // Environment from docker .ENV
+// @TODO fix ENV
+const ENV = require('../.env.json');
 
+/*
+Topic1Name=kafka-test-topic
+Topic1Partition=1
+Topic1Replicas=1
+ZookeeperPort=2181
+KafkaProtocol=kafka
+KafkaPort=9092
+KafkaIP=localhost
+Throttle=250
+ */
 /**
  * KafkaConfig
  */
@@ -8,7 +20,7 @@ module.exports = {
     client: {
       // debug: 'all',
       'group.id': 'kafka',
-      'metadata.broker.list': `${ENV.KafkaIP}:${ENV.KafkaPort}`,
+      'metadata.broker.list': 'localhost:9092',
       'enable.auto.commit': false
     },
     topics: [ENV.Topic1Name],

src/index.d.ts

@@ -166,6 +166,12 @@ export interface MessagePayload {
 }
 
 export interface DeliveryReport {
+  topic: string;
+  partition: number;
+  offset: number;
+  key: string;
+  opaque: string;
+  size: number;
 
 }
 

src/producer.js

@@ -62,6 +62,7 @@ class Producer extends KafkaClient {
    * @param {String} [key=null] - keyed message (optional)
    * @param {String} [opaque=null] - opaque token which gets passed along to your delivery reports
    * @return {Promise<DeliveryReport>}
+   * @TODO will delivery report be synchronized with produce?
    */
   publish(message, topic = this._config.topics[0], partition = -1, key = null, opaque = null) {
     return new Promise((resolve, reject) => {

test.js

@@ -10,7 +10,11 @@ describe('Kafka  Test', () => {
     });
 
     afterEach(() => {
-      return Promise.all([producer.disconnect(), consumer.disconnect()]);
+      console.log('Disconnect');
+      return Promise.all([producer.disconnect(), consumer.disconnect()])
+        .then((data) => {
+          console.log('Disconnected', data);
+        });
     });
 
     it('Pub Sub', () => {