more relay stuff, setting up server

1 files changed, 43 insertions, 10 deletions

diff --git a/app/relay/index.js b/app/relay/index.js
index 22dc85f..2ed8e10 100644
--- a/app/relay/index.js
+++ b/app/relay/index.js
@@ -1,6 +1,33 @@
-const zerorpc = require('zerorpc')
 require('dotenv').config()
 
+const io = require('socket.io-client')
+const ss = require('socket.io-stream')
+const zerorpc = require('zerorpc')
+const Readable = require('stream').Readable;
+
+let remote = io.connect(process.env.SOCKETIO_REMOTE);
+
+remote.on('cmd', (data) => {
+  if (! data.cmd || ! data.payload) {
+    console.log('malformed param...?')
+    return
+  }
+  console.log('got', data.cmd)
+  switch (data.cmd) {
+    case 'send_param':
+      rpc.invoke(data.cmd, data.payload.key, data.payload.value, (err, res, more) => {
+        console.log('sent param', res)
+      })
+      break
+    case 'get_params':
+      rpc.invoke(data.cmd, (err, res, more) => {
+        console.log('got params', res)
+        remote.emit('params', res)
+      })
+      break
+  }
+})
+
 let rpc = new zerorpc.Client()
 rpc.connect("tcp://127.0.0.1:" + process.env.RPC_PORT)
 rpc.on("error", function(error) {
@@ -9,17 +36,24 @@ rpc.on("error", function(error) {
 console.log('RPC listening on port ' + process.env.RPC_PORT)
 
 let relay = new zerorpc.Server({
-  send_file: function (file, reply) {
-    reply()
-    console.log('got file, ' + file.length + ' bytes')
-  },
+  // Called when the worker starts up and is ready to receive params.
   connected: function (msg, reply) {
     reply()
-    console.log("got connect, trying to call back.")
-    rpc.invoke("send_param", "foo", "bar", (err, res, more) => {
-      console.log('sent param')
-    })
+    console.log("got connect from " + msg)
     return true
+  },
+
+  send_frame: function (fn, frame, reply) {
+    reply()
+    console.log('got frame, ' + frame.length + ' bytes')
+    var stream = ss.createStream();
+    ss(remote).emit('frame', stream, {name: fn})
+
+    var s = new Readable();
+    s._read = function noop() {}
+    s.push(frame)
+    s.push(null);
+    s.pipe(stream)
   }
 })
 relay.on("error", function(error) {
@@ -27,4 +61,3 @@ relay.on("error", function(error) {
 })
 relay.bind("tcp://0.0.0.0:" + process.env.RELAY_PORT);
 console.log('Relay listening on port ' + process.env.RELAY_PORT)
-