Merge pull request #8 from sjwoodman/demo-prep

Demo prep

Author: sjwoodman

cleanup.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+
+oc delete dc -l app=${1}
+oc delete cm -l streamzi.io/kind=ev
+oc delete cm -l strimzi.io/kind=topic

deploy-all.sh

@@ -1,7 +1,14 @@
 #!/usr/bin/env bash
 
 mvn -f ./pom.xml clean install
-./manager/build.sh;
-./operations/log-data/build.sh;
-./operations/nodejs/filter-events/build.sh;
-./operations/random-data/build.sh;
+cd manager
+./build.sh;
+cd ../operations/random-data/
+./build.sh
+cd ../log-data/
+./build.sh
+cd ../nodejs/filter-events/
+./build.sh
+
+cd ../../
+

manager/src/main/java/io/streamzi/openshift/API.java

@@ -31,7 +31,7 @@ public class API {
     @EJB(beanInterface = ClientContainer.class)
     private ClientContainer container;
 
-    private final String bootstrapServersDefault = "my-cluster-kafka";
+    private final String bootstrapServersDefault = "my-cluster-kafka:9092";
 
     @GET
     @Path("/pods")

manager/src/main/java/io/streamzi/openshift/ProcessorFlowDeployer.java

@@ -128,6 +128,7 @@ private void populateTopicMaps(Map<String, DeploymentConfig> deploymentConfigs)
         final ConfigMap cm = new ConfigMap();
         String topicName;
 
+        //todo: deal with unconnected outputs - should still have a topic created
         for (ProcessorLink link : flow.getLinks()) {
             logger.info("Processing link");
             topicName = link.getSource().getParent().getUuid() + "-" + link.getSource().getName();

operations/nodejs/filter-events/main.js

@@ -3,70 +3,79 @@
  * To change this template file, choose Tools | Templates
  * and open the template in the editor.
  */
-console.log("Starting KAFKA Processor in 10 seconds");
-setTimeout(function () {
-    var kafkaHost = process.env.STREAMZI_KAFKA_BOOTSTRAP_SERVER;
-    var nodeUuid = process.env.STREAMZI_NODE_UUID;
-    var sourceTopic = process.env.INPUT_DATA;
-    var targetTopic = process.env.OUTPUT_DATA;
-    var threshold = process.env.THRESHOLD;
-
-    if(threshold === undefined){
-        threshold = 0.5
-    }
 
-    console.log("Host: " + kafkaHost);
-    console.log("Input topic: " + sourceTopic);
-    console.log("Output topic: " + targetTopic);
-    console.log("Threshold: " + threshold);
+var kafkaHost = process.env.STREAMZI_KAFKA_BOOTSTRAP_SERVER;
+var nodeUuid = process.env.STREAMZI_NODE_UUID;
+var sourceTopic = process.env.INPUT_DATA;
+var targetTopic = process.env.OUTPUT_DATA;
+var threshold = process.env.THRESHOLD;
+
+if(threshold === undefined){
+    threshold = 0.5
+}
+
+console.log("Host: " + kafkaHost);
+console.log("Input topic: " + sourceTopic);
+console.log("Output topic: " + targetTopic);
+console.log("Threshold: " + threshold);
+
+var kafka = require("kafka-node");
+var Producer = kafka.Producer;
+var Consumer = kafka.Consumer;
+const client = new kafka.KafkaClient({
+    kafkaHost: kafkaHost
+});
+
+
+const producer = new Producer(client);
+producer.on("ready", function(){
+    console.log("Producer ready");
+});
+
+producer.on("error", function(err){
+    console.log("Producer error: " + err);
+});
 
-    var kafka = require("kafka-node");    
-    var Producer = kafka.Producer;    
-    var Consumer = kafka.Consumer;
-    const client = new kafka.KafkaClient({
-        kafkaHost: kafkaHost
-    });
-    
-    
-    const producer = new Producer(client);    
-    producer.on("ready", function(){
-        console.log("Producer ready");
-    });
-    
-    producer.on("error", function(err){
-        console.log("Producer error: " + err);
-    });
-    
-
-    
-    const consumer = new Consumer(
-            client,
-            [
-                {
-                    topic: sourceTopic
-                }
-            ], {
-        autoCommit: true,
-        groupId: nodeUuid
-    });
 
-    consumer.on("message", function (message) {
-        console.log("y");
-        var value = message.value;
-        var cloudEvent = JSON.parse(value);
-        var numberValue = cloudEvent.data.value;
-        console.log(numberValue);
-        
-        if(numberValue>threshold){
-
-            var payloads = [
-                {
-                    topic: targetTopic,
-                    messages: message.value
-                }
-            ];
-            producer.send(payloads, function(err, data){
-            });
-        }
+
+const consumer = new Consumer(
+        client,
+        [
+            {
+                topic: sourceTopic,
+                groupId: nodeUuid
+            }
+        ], {
+    autoCommit: true,
+    groupId: nodeUuid
+});
+
+consumer.on("ready", function(){
+    console.log("Consumer ready");
+});
+
+ consumer.on("error", function(err){
+        console.log("Consumer error: " + err);
     });
-}, 10000);
+
+
+consumer.on("message", function (message) {
+
+    var value = message.value;
+    var cloudEvent = JSON.parse(value);
+    var numberValue = cloudEvent.data.value;
+    console.log(numberValue);
+
+    if(numberValue>threshold){
+
+        var payloads = [
+            {
+                topic: targetTopic,
+                messages: message.value
+            }
+        ];
+        producer.send(payloads, function(err, data){
+        });
+    }
+});
+