moving internal pub/sub to kafka

initial move complete for internal pub/sub to Kafka - needs testing

dispatch_module/constants.json

@@ -2,5 +2,5 @@
     "mqtt_url": "10.129.6.5",
     "registry_url" :"10.129.6.5:5000/",
     "master_port": 8080,
-    "master_address": "10.129.6.5"
+    "master_address": "localhost"
 }
\ No newline at end of file

dispatch_module/dispatch_daemon/config.json

+{
+    "id": "tpt8hqn7ok"
+}
\ No newline at end of file

dispatch_module/dispatch_daemon/index.js

 const mqtt = require('mqtt')
 const constants = require(".././constants.json")
-console.log(constants.mqtt_url);
-
-const client = mqtt.connect('mqtt://' + constants.mqtt_url)
+const node_id = require("./config.json").id
 const libSupport = require('./lib')
 const execute = require('./execute')
 const fs = require('fs')
 
-const node_id = libSupport.makeid(10)
 const local_repository = __dirname + "/local_repository/"
 const host_url = "http://" + constants.master_address + ":" + constants.master_port
-client.on('connect', function () {
-    client.subscribe(node_id, function (err) {
-        if (!err) {
-            console.log("node listening to id", node_id);
-            
-        }
-    })
-})
 
-client.on('message', function (topic, message) {
+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: node_id, partition: 0, offset: 0}
+        ],
+        [
+            { autoCommit: true }
+        ])
 
+consumer.on('message', function (message) {
+    console.log(message);
+    let topic = message.topic
+        message = message.value
     message = JSON.parse(message)
     if (topic !== 'heartbeat') {
         let runtime = message.runtime
@@ -34,30 +42,43 @@ client.on('message', function (topic, message) {
                     
                     if (runtime === "isolate")
                         execute.runIsolate(local_repository + functionHash).then(result => {
-                            client.publish("response", JSON.stringify({
+                                producer.send([{
+                                    topic: "response", 
+                                    messages: JSON.stringify({
                                         status: "success",
                                         result,
-                                function_id
-                            }))
+                                        function_id})
+                                }], () =>{})
                         })
                     else if (runtime === "process")
                         execute.runProcess(local_repository + functionHash).then(result => {
-                            client.publish("response", JSON.stringify({
+                                producer.send(
+                                [{
+                                    topic: "response", 
+                                    messages: JSON.stringify({
                                         status: "success",
                                         result,
-                                function_id
-                            }))
+                                        function_id})
+                                }], () =>{})
                         })
                     else if (runtime === "container")
                         execute.runContainer(functionHash).then(result => {
-                            client.publish("response", JSON.stringify({
+                                producer.send(
+                                [{
+                                    topic: "response", 
+                                    messages: JSON.stringify({
                                         status: "success",
                                         result,
-                                function_id
-                            }))
+                                        function_id})
+                                }], () =>{})
                         })
                     else {
-                        client.publish("response", JSON.stringify({ status: "unknown runtime" }))
+                            producer.send(
+                            [{
+                                topic: "response", 
+                                messages: JSON.stringify({ status: "unknown runtime" })
+                            }], () =>{})
+                        
                         return
                     }
                 })
@@ -65,30 +86,43 @@ client.on('message', function (topic, message) {
                 
                 if (runtime === "isolate")
                     execute.runIsolate(local_repository + functionHash).then(result => {
-                        client.publish("response", JSON.stringify({
+                            producer.send(
+                            [{
+                                topic: "response", 
+                                messages: JSON.stringify({
                                     status: "success",
                                     result,
-                            function_id
-                        }))
+                                    function_id})
+                            }], () =>{})
                     })
                 else if (runtime === "process")
                     execute.runProcess(local_repository + functionHash).then(result => {
-                        client.publish("response", JSON.stringify({
+                            producer.send(
+                            [{
+                                topic: "response", 
+                                messages: JSON.stringify({
                                     status: "success",
                                     result,
-                            function_id
-                        }))
+                                    function_id})
+                            }], () =>{})
                     })
                 else if (runtime === "container")
                     execute.runContainer(functionHash).then(result => {
-                        client.publish("response", JSON.stringify({
+                            producer.send(
+                            [{
+                                topic: "response", 
+                                messages: JSON.stringify({
                                     status: "success",
                                     result,
-                            function_id
-                        }))
+                                    function_id})
+                            }], () =>{})
                     })
                 else {
-                    client.publish("response", JSON.stringify({ status: "unknown runtime" }))
+                        producer.send(
+                        [{
+                            topic: "response", 
+                            messages: JSON.stringify({ status: "unknown runtime" })
+                        }], () =>{})
                     return
                 }
             }
@@ -99,7 +133,13 @@ client.on('message', function (topic, message) {
 })
 
 function heartbeat() {
-    client.publish("heartbeat", JSON.stringify({"address": node_id}))
+    let payload = [{
+        topic: "heartbeat",
+        messages: JSON.stringify({"address": node_id})
+    }]
+    producer.send(payload, function() {
+
+    })
 }
 
 setInterval(heartbeat, 1000);
\ No newline at end of file

dispatch_module/dispatch_daemon/package.json

@@ -14,6 +14,7 @@
     "express": "^4.17.1",
     "express-fileupload": "^1.1.6",
     "isolated-vm": "^3.0.0",
+    "kafka-node": "^5.0.0",
     "morgan": "^1.9.1",
     "mqtt": "^3.0.0",
     "redis": "^2.8.0"

dispatch_module/dispatcher/index.js

@@ -26,8 +26,7 @@ let kafka = require('kafka-node'),
             { topic: 'response', partition: 0, offset: 0}, { topic: 'heartbeat' }
         ],
         [
-            { autoCommit: false },
-            { fromOffset: true}
+            { autoCommit: true }
         ])
 
 let db = new Map()
@@ -161,11 +160,16 @@ function dispatch() {
         let function_id = libSupport.makeid(20)
         console.log("Dispatching function with Id", function_id, runtime);
         let node_id = getAddress()
-        client.publish(node_id, JSON.stringify({
+        let payload = [{
+            topic: node_id,
+            messages: JSON.stringify({
                 "type": "execute",
                 function_id,
                 runtime, functionHash
-        }))
+            }),
+            partition: 0
+        }]
+        producer.send(payload, () => {})
         db.set(function_id, res)
         
     }
@@ -204,7 +208,25 @@ app.listen(port, () => console.log(`Server listening on port ${port}!`))
 // })
 
 consumer.on('message', function (message) {
+    // console.log(message);
+    let topic = message.topic
+        message = message.value
+    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);
+        }
+    }
 });
 
-setInterval(dispatch, 1000);
+setInterval(dispatch, 2000);

dispatch_module/dispatcher/package.json

@@ -14,6 +14,7 @@
     "express": "^4.17.1",
     "express-fileupload": "^1.1.6",
     "isolated-vm": "^3.0.0",
+    "kafka-node": "^5.0.0",
     "morgan": "^1.9.1",
     "mqtt": "^3.0.0",
     "nano": "^8.1.0",