update

cli/src/commands/get-stream.js

+// @flow
+const path = require("path");
+const chalk = require("chalk");
+const fs = require("fs");
+const util = require("util");
+const net = require("net");
+const process = require("process");
+var request = require("request");
+var csv = require("csv-stream");
+const fetch = require("node-fetch");
+
+const handleErrors = require("../utils/handleErrors");
+const getMaster = require("../utils/getMaster");
+
+module.exports.command = "put-stream";
+module.exports.describe = "Writes localfilename to the SDFS as sdfsfilename.";
+module.exports.builder = (yargs: any) => yargs;
+
+
+
+module.exports.handler = handleErrors(async (argv: {}) => {
+  const master = await getMaster();
+  getCSVdata();
+
+  setInterval(function sync() {
+    // if (queue.length > 0) {
+    //   const body = queue.pop();
+      var target = "http://"+ master + ":3000/sink/";
+    //   console.log(target)
+    //   console.log(JSON.stringify(body).state)
+      fetch(target)
+        .then(res => console.log(res.text()))
+        // .then(json => {
+        // //   if (json.action !== "END") {
+        // //     queue.push(json);
+        // //   } else {
+        //     // console.log(json);
+        // //   }
+        // });
+    // }
+  
+  }, 500);
+  
+});

cli/src/commands/put-stream.js

@@ -40,7 +40,7 @@ async function getCSVdata() {
       if (messages.length == 10000) {
         allMessages.push(messages);
         const body = { action: "START", state: messages[0] };
-        console.log(body)
+        // console.log(body)
         queue.push(body);
         messages = [];
       }
@@ -78,7 +78,9 @@ module.exports.handler = handleErrors(async (argv: {}) => {
             // console.log(json);
         //   }
         });
-    }
+    } else {
+        process.exit()
+    } 
   
   }, 500);
   

sfds/index.js

@@ -408,6 +408,10 @@ app.post("/spout", function(req, res) {
   res.send("got it");
 });
 
+app.post("/sink", function(req, res) {
+  // console.log("bolt state: " +req.body.state)
+  res.send("IM A SINK");
+});
 
 
 round_robin = 0