kafka node integration started

started listener for kafka node in dispatcher, but not working correctly

design_decisions.txt

+Design decisions to be considered:
+1. Container: copy executable file inside prebuilt images.
+2. Container: run warm request based container server instead of current arch
+	possibly will require request redirection.
+3. Move away from mosquitto to Apache Kafka / Zookeeper
+4. Implement heartbeat based daemon monitor.

dispatcher/index.js

@@ -8,10 +8,28 @@ const fs = require('fs')
 const { spawn } = require('child_process');
 const morgan = require('morgan')
 const mqtt = require('mqtt')
-const client = mqtt.connect('mqtt://' + constants.mqtt_url)
+// const client = mqtt.connect('mqtt://' + constants.mqtt_url)
 const app = express()
 const libSupport = require('./lib')
 
+let kafka = require('kafka-node'),
+    Producer = kafka.Producer,
+    KeyedMessage = kafka.KeyedMessage,
+    client = new kafka.KafkaClient({ 
+        kafkaHost: '10.129.6.5:9092',
+        autoConnect: true
+    }),
+    producer = new Producer(client),
+    Consumer = kafka.Consumer,
+    consumer = new Consumer(client,
+        [
+            { topic: 'response', partition: 0, offset: 0}, { topic: 'heartbeat' }
+        ],
+        [
+            { autoCommit: false },
+            { fromOffset: true}
+        ])
+
 let db = new Map()
 app.use(morgan('combined'))
 app.use(bodyParser.urlencoded({ extended: true }))
@@ -158,31 +176,35 @@ function getAddress() {
 }
 
 
-client.on('message', function (topic, message) {
-
-    if (topic === "response") {
-        message = JSON.parse(message)
-        console.log(message);
-        let res = db.get(message.function_id)
-        res.json({
-            "status": "success",
-            "reply": message.result
-        })
-        db.delete(message.function_id)
-    } else if (topic === "heartbeat") {
-        message = JSON.parse(message)
-        if (workerNodes.indexOf(message.address) === -1)  {
-            workerNodes.push(message.address)
-            console.log(workerNodes);
-	}
-    }
-})
+// client.on('message', function (topic, message) {
+
+//     if (topic === "response") {
+//         message = JSON.parse(message)
+//         console.log(message);
+//         let res = db.get(message.function_id)
+//         res.json({
+//             "status": "success",
+//             "reply": message.result
+//         })
+//         db.delete(message.function_id)
+//     } else if (topic === "heartbeat") {
+//         message = JSON.parse(message)
+//         if (workerNodes.indexOf(message.address) === -1)  {
+//             workerNodes.push(message.address)
+//             console.log(workerNodes);
+// 	}
+//     }
+// })
 
 app.listen(port, () => console.log(`Server listening on port ${port}!`))
 
-client.on('connect', function () {
-    client.subscribe("response")
-    client.subscribe("heartbeat")
-})
+// client.on('connect', function () {
+//     client.subscribe("response")
+//     client.subscribe("heartbeat")
+// })
+
+consumer.on('message', function (message) {
+    console.log(message);
+});
 
 setInterval(dispatch, 1000);