require('dotenv').config() const io = require('socket.io-client') const zerorpc = require('zerorpc') const Readable = require('stream').Readable const runner = require('./runner') let remote, relay, rpc, rpc_connected = false remote = io.connect(process.env.SOCKETIO_REMOTE) remote.on('cmd', (data) => { console.log('cmd data', data) if (! data.cmd) return console.log('malformed param...?') console.log('got', data.cmd) switch (data.cmd) { case 'set_param': if (! data.payload) return rpc.invoke(data.cmd, data.payload.key, data.payload.value, (err, res, more) => { console.log('sent param, got response', res) }) break case 'get_params': case 'get_last_frame': rpc.invoke(data.cmd, (err, res, more) => { console.log('got params', res) remote.emit('res', { cmd: data.cmd, res: res, }) }) break default: rpc.invoke('send_command', data.cmd, data.payload || null, (err, res, more) => { console.log('sent command', res) remote.emit('res', { cmd: data.cmd, res: res, }) }) break } }) remote.on('system', (data) => { console.log('system:', data.cmd) switch(data.cmd) { case 'run_system_command': runner.run_system_command(data.payload, (error, stdout, stderr) => { remote.emit('system_res', { type: 'command_output', cmd: data.payload, error, stdout, stderr }) }) break case 'get_status': remote.emit('system_res', { type: 'relay_status', rpc_connected: rpc_connected, runner: runner.status(), }) break default: remote.emit('system_res', { error: 'unknown system command' }) break } }) rpc = new zerorpc.Client() rpc.connect('tcp://127.0.0.1:' + process.env.RPC_PORT) rpc.on('error', function(error) { console.error('RPC server error:', error) }) console.log('RPC listening on port ' + process.env.RPC_PORT) relay = new zerorpc.Server({ // Called when the worker starts up and is ready to receive params. connected: function(msg, reply) { reply() console.log('got connect from ' + msg) remote.emit('system_res', { type: 'rpc_connected', runner: runner.status(), }) rpc_connected = true return true }, send_frame: function(fn, meta, frame, reply) { reply() // console.log('got frame, ' + frame.length + ' bytes') remote.emit('frame', { fn: fn, meta: meta, frame: frame }) }, send_status: function(key, value, reply) { reply() remote.emit('status', { key: key, value: value }) }, disconnecting: function(){ reply() remote.emit('system_res', { type: 'rpc_disconnected', }) rpc_connected = false return true }, }) relay.on('error', function(error) { console.error('Relay server error:', error) }) relay.bind('tcp://0.0.0.0:' + process.env.RELAY_PORT) console.log('Relay listening on port ' + process.env.RELAY_PORT) rpc.invoke('ping', (err, res, more) => { console.log('sent ping', res) if (res === 'pong') { remote.emit('system_res', { type: 'rpc_connected', runner: runner.status() }) rpc_connected = true } else { remote.emit('system_res', { type: 'rpc_disconnected', }) rpc_connected = false } }) module.exports = { relay: relay, remote: remote, }