diff --git a/client.js b/client.js index c2a04bc3..34a55305 100644 --- a/client.js +++ b/client.js @@ -12,6 +12,8 @@ const _ = require('underscore'); const deepmerge = require("deepmerge"); const getMac = require('getmac').getMac; const rmdirSync = require('rmdir-sync'); +const ioClient = require("socket.io-client"); + // internal libraries const objectOps = require("./lib/objectOps.js"); const fileOps = require("./lib/fileOps.js"); @@ -415,6 +417,15 @@ write-data=__PATH__executable__/../../../instances/" + instance + "\r\n\ // ensure instancemanagement only ever runs once _.once(instanceManagement); function instanceManagement() { + /* Open websocket connection to master */ + var socket = ioClient(config.masterIP+":"+config.masterPort); + socket.on("hello", function(){ + socket.emit("registerSlave", { + instanceID: config.instanceID, + }); + setInterval(B=> socket.emit("heartbeat"), 10000); + }); + console.log("Started instanceManagement();"); // load plugins and execute onLoad event let pluginDirectories = fileOps.getDirectoriesSync("./sharedPlugins/"); @@ -790,6 +801,7 @@ function instanceManagement() { console.log("Unable to get JSON master/api/inventory, master might be unaccessible"); } else if (response && response.body) { // Take the inventory we (hopefully) got and turn it into the format LUA accepts + if(Buffer.isBuffer(response.body)) {console.log(response.body.toString("utf-8")); throw new Error();} // We are probably contacting the wrong webserver try { var inventory = JSON.parse(response.body); } catch (e){ @@ -808,56 +820,54 @@ function instanceManagement() { } }); }, 1000); - // REMOTE SIGNALLING - // send any signals the slave has been told to send - setInterval(function () { - // Fetch combinator signals from the server - needle.post(config.masterIP + ":" + config.masterPort + '/api/readSignal', { - since: lastSignalCheck - }, function (err, response, body) { - if (response && response.body) { - if(typeof response.body == "string") response.body = JSON.parse(response.body); - // Take the new combinator frames and compress them so we can use a single command - let frameset = []; - for (let i = 0; i < response.body.length; i++) { - frameset[i] = response.body[i].frame; - } - // console.log(frameset); - // Send all our compressed frames - messageInterface("/silent-command remote.call('clusterio', 'receiveMany', '" + JSON.stringify(frameset) + "')"); - } - }); - // after fetching all the latest frames, we take a timestamp. During the next iteration, we fetch all frames submitted after this. - lastSignalCheck = Date.now(); - - // get outbound frames from file and send to master - // get array of lines in file, each line should correspond to a JSON encoded frame - let signals = fs.readFileSync(instancedirectory + "/script-output/txbuffer.txt", "utf8").split("\n"); - // if we actually got anything from the file, proceed and reset file - if (signals[0]) { - fs.writeFileSync(instancedirectory + "/script-output/txbuffer.txt", ""); - // loop through all our frames - for (let i = 0; i < signals.length; i++) { - (function (i) { - if (signals[i]) { - // signals[i] is a JSON array called a "frame" of signals. We timestamp it for storage on master - // then we unpack and RCON in this.frame to the game later. - let framepart = JSON.parse(signals[i]); - let doneframe = { - time: Date.now(), - frame: framepart, // thats our array of objects(single signals); - } - // console.log(doneframe); - needle.post(config.masterIP + ":" + config.masterPort + '/api/setSignal', doneframe, function (err, response, body) { - if (response && response.body) { - // In the future we might be interested in whether or not we actually manage to send it, but honestly I don't care. - } - }); - } - })(i); - } + /* REMOTE SIGNALLING + * send any signals the slave has been told to send + * Fetch combinator signals from the server + */ + socket.on("processCombinatorSignal", circuitFrameWithMeta => { + if(circuitFrameWithMeta && typeof circuitFrameWithMeta == "object" && circuitFrameWithMeta.frame && Array.isArray(circuitFrameWithMeta.frame) && objectOps.isJSON(circuitFrameWithMeta.frame)){ + messageInterface("/silent-command remote.call('clusterio', 'recieveMany', '"+JSON.stringify(circuitFrameWithMeta.frame)+"')"); } - }, 1000); + }); + // get outbound frames from file and send to master + // get array of lines in file, each line should correspond to a JSON encoded frame + let signals = fs.readFileSync(instancedirectory + "/script-output/txbuffer.txt", "utf8").split("\n"); + // if we actually got anything from the file, proceed and reset file + let readingTxBufferSoon = false; + let txBufferClearCounter = 0; + fs.watch(instancedirectory + "/script-output/txbuffer.txt", "utf-8", (eventType, filename) => { + if(!readingTxBufferSoon){ // use a 100ms delay to avoid messing with rapid sequential writes from factorio (I think that might be a problem maybe?) + readingTxBufferSoon = true; + setTimeout(()=>{ + txBufferClearCounter++; + fs.readFile(instancedirectory + "/script-output/txbuffer.txt", "utf-8", (err, signals) => { + if(txBufferClearCounter > 500){ + fs.writeFileSync(instancedirectory + "/script-output/txbuffer.txt", ""); + txBufferClearCounter = 0; + } + console.log(signals); + if (signals[0]) { + fs.writeFileSync(instancedirectory + "/script-output/txbuffer.txt", ""); + // loop through all our frames + for (let i = 0; i < signals.length; i++) { + if (signals[i]) { + // signals[i] is a JSON array called a "frame" of signals. We timestamp it for storage on master + // then we unpack and RCON in this.frame to the game later. + let framepart = JSON.parse(signals[i]); + let doneframe = { + time: Date.now(), + frame: framepart, // thats our array of objects(single signals); + } + // send to master using socket.io, opened at the top of instanceManagement() + socket.emit("combinatorSignal", doneframe); + } + } + } + }); + readingTxBufferSoon = false; + },100); + } + }); } // END OF INSTANCE START --------------------------------------------------------------------- // string, function diff --git a/master.js b/master.js index 695d3f9c..f667f27a 100644 --- a/master.js +++ b/master.js @@ -671,23 +671,45 @@ class mapRequester { }); } } +/* Websockets for send and recieve combinators */ +var wsSlaves = {}; +class wsSlave { + constructor(instanceID, socket){ + this.instanceID = instanceID; + this.socket = socket; + this.lastBeat = Date.now(); + + this.socket.on("heartbeat", () => { + this.lastBeat = Date.now(); + }); + this.socket.on("combinatorSignal", circuitFrameWithMeta => { + if(circuitFrameWithMeta && typeof circuitFrameWithMeta == "object"){ + Object.keys(wsSlaves).forEach(instanceID => { + wsSlaves[instanceID].socket.emit("processCombinatorSignal", circuitFrameWithMeta); + }); + } + }); + } +} io.on('connection', function (socket) { // cleanup dead sockets from disconnected people - Object.keys(mapRequesters).forEach(requesterName => { - let requester = mapRequesters[requesterName]; - if(requester.lastBeat < (Date.now() - 30000)){ - console.log("remoteMap | There are currently "+Object.keys(mapRequesters).length+" map requesters, deleting one on timeout"); - delete mapRequesters[requesterName]; // we have to use the full name here or we will only kill the pointer - } - }); - Object.keys(slaveMappers).forEach(mapperName => { - let mapper = slaveMappers[mapperName]; - if(mapper.lastBeat < (Date.now() - 30000)){ - console.log("remoteMap | There are currently "+Object.keys(slaveMappers).length+" mappers, deleting one on timeout"); - delete slaveMappers[mapperName]; // we have to use the full name here or we will only kill the pointer - } + let terminatedConnections = 0; + let currentConnections = Object.keys(mapRequesters).length + Object.keys(slaveMappers).length + Object.keys(wsSlaves).length; + [mapRequesters, slaveMappers, wsSlaves].forEach(list => { + Object.keys(list).forEach(connectionID => { + let connection = list[connectionID]; + if(connection.lastBeat < (Date.now() - 30000)){ + terminatedConnections++; + delete list[connectionID]; + } + }); }); + if(terminatedConnections > 0) console.log("SOCKET | There are currently "+currentConnections+" websocket connections, deleting "+terminatedConnections+" on timeout"); + + // tell our friend that we are listening socket.emit('hello', { hello: 'world' }); + + /* initial processing for remoteMap */ socket.on('registerSlaveMapper', function (data) { slaveMappers[data.instanceID] = new slaveMapper(data.instanceID, socket); console.log("remoteMap | SOCKET registered map provider for "+data.instanceID); @@ -699,6 +721,15 @@ io.on('connection', function (socket) { socket.emit("mapRequesterReady", true); console.log("remoteMap | SOCKET registered map requester for "+data.instanceID); }); + + /* Websockets for send and recieve combinators */ + socket.on("registerSlave", function(data) { + if(data && data.instanceID){ + wsSlaves[data.instanceID] = new wsSlave(data.instanceID, socket); + console.log("SOCKET | Created new wsSlave: "+ data.instanceID); + } + }); + }); module.exports = app;