aperx/factorio-clusterio
1
0 Fork 0
mirror of https://github.com/clusterio/clusterio.git synced 2026-11-13 23:14:53 +09:00
Code Issues Packages Projects Releases Wiki Activity

Daemonize slave

Browse Source 
Rework client.js to support running multiple instances at once and be
managed remotely and implement the neccessary communication
infrastructure for this.

Due to circular dependencies on the implementation of this the master
branch is significantly broken after this.
This commit is contained in:
Hornwitser committed 2019-11-23 15:16:47 +01:00
1 parent 5858d97857
commit 7a2abaa5db
29 files changed
+2811 -800
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitignore
+2
View File
@@ -34,6 +34,8 @@ Factorio*
/database
database/items.db
config.json
config-control.json
config-slave.json
package-lock.json
secret-api-token.txt
mod_cache.json
CHANGELOG.md
+21
View File
@@ -12,6 +12,10 @@ Version 2.0.0
event_loader lib into the savegame. Regular freeplay games can now be
used with Clusterio and will be compatible without having to convert them
to Hotpatch.
- Daemonized slaves. Slaves now have the ability to run multiple Factorio
instances, starting and stopping them individually. To manage this the
local command line interface has been replaced with a remote interface on
the master server that can be accessed through the clusterctl cli tool.
### Changes
@@ -50,6 +54,10 @@ Version 2.0.0
- Removed factorio_version from config. The version installed is auto
detected and used instead.
- Removed the playerManager specific command CLI tools/delete_player.js.
- Creating an instance, creating a save for an instance and starting an
instance is now three separate commands.
- Removed oddball limits to slaves.json size.
- Moved slave specific configuration into its own configuration file.
### Breaking Changes
@@ -74,6 +82,19 @@ Version 2.0.0
the old unmaintained and no longer needed factorioClusterioClient.
- Removed the remoteCommands plugin and the old runCommand interface. Breaks
playerManager and external tools sending commands.
- Removed broken serverManager plugin.
- Removed fields info, time, rconPort, rconPassword, serverPort, unique,
mods, instanceName, playerCount, mac, and meta from the slaves in the
slave database.
- Removed getInstanceName and lib/clusterTools. Breaks playerManager,
and discordChat.
- Removed the /api/rconPasswords, and /api/slaves endpoints. Breaks web
interface, trainTeleports, and discordChat.
- Removed the /api/getSlaveMeta and /api/editSlaveMeta endpoints. Breaks
researchSync, and UPSdisplay.
- Removed the hello event from the socket connection handshake. Breaks
playerManager, trainTeleports, serverSelect, and discordChat
- Changed the format of database/slaves.json.
Version 1.2.3
Dockerfile
+2
View File
@@ -1,3 +1,5 @@
# Warning: This is currently broken as a result of
# changes done in the 2.0 refactoring.
FROM node:10
RUN apt-get update && apt install git curl tar -y
RUN mkdir factorioClusterio
README.md
+75 -79
View File
@@ -10,6 +10,8 @@ Discord for development/support/play: https://discord.gg/5XuDkje
## Important notice
**WARNING: The master branch is currently broken, do not use it!**
This is the development branch for factorioClusterio 2.0 which is currently undergoing heavy
restructuring and refactoring. Expect plugins and existing installations to frequently break when
using this branch. If you don't want to be an alpha tester for 2.0 please use the stable
@@ -30,17 +32,15 @@ the [1.2.x branch][1.2.x] for instructions on how to install the stable version.
### Table of contents
* [Introduction & methodology](#introduction)
* [Ubuntu setup](#ubuntu-setup)
* [Windows setup](#windows-setup)
* [Running Clusterio](#running-clusterio)
* [Master Server](#master-server)
* [Slaves](#slaves)
* [Instances](#instances)
* [Optional plugins](#Plugins)
* [Common problems](#Common-problems)
* [Command cheatsheet](#cheatsheet)
## Introduction
Features:
@@ -123,39 +123,15 @@ Optional step (if you want to use pm2):
Now you need to edit the `config.json` file. If you skip this step nothing will work.
Pretty much all the blank fields should be filled in, except on the master where a few can be omitted.
* You get the `masterAuthToken` from `secret-api-token.txt` in the master install dir after running the master twice.
* You get your factorio matchmaking token from factorio.com
* The `masterAuthSecret` should never be touched unless you want to invalidate everyones authentication tokens
**Master**
node master
Or with pm2 (it's recommened to run it without pm2 first):
pm2 start master --name master
**Server Host**
To download the mod for all its non vanilla features and items, (optional, but very recommended)
node client manage shared mods add clusterio
To create a new instance (its own save, set of mods and config files)
node client start [instancename]
To launch an instance with pm2
pm2 start --name [instancename] client -- start [instancename]
use `nano config.json` to change settings.
**Ubuntu with Docker**
The Docker support for Clusterio is curently broken. If you need it
open an issue about it.
<!--
Clusterio has *very* limited support for using docker.
sudo docker build -t clusterio --no-cache --force-rm factorioClusterio
@@ -165,6 +141,7 @@ Clusterio has *very* limited support for using docker.
sudo docker run --name slave -e MODE=client -e INSTANCE=world1 -v /srv/clusterio/instances:/factorioClusterio/instances -p 1235:34167 -it --restart=unless-stopped danielvestol/clusterio
The -v flag is used to specify the instance directory. Your instances (save files etc) will be stored there.
-->
## Windows setup
@@ -219,35 +196,83 @@ reboot when you are done, then proceed to the next steps. *reboots matter*
4. Run `node master` to generate the athentication token into secret-api-token.txt
5. Run `node master` again to start the master server.
**Server Host**
1. Do step 1 and 2 of the Master section above *OR* use the same folder that was created in that section.
2. Open `config.json` with a text editor and configure as desired. You will need to set masterAuthToken to string found in secret-api-token.txt on the master server.
3. Optionally run the command `node client manage shared mods add clusterio` to add the clusterio mod (needed for item teleports.)
4. Run `node client start [instancename]` to create a new instance. Repeat it again to start the instance.
To connect to a master server running on a remote machine, open config.json with your favourite text editor (notepad++). You can also set it up to use the official server browser.
Change `masterURL `to something like `http://203.0.113.33:8080` (provided by master server owner)
Change `masterAuthToken` to the value found in `secret-api-token.txt` on the master server
Repeat step 4 for more servers on one machine. You should be able to find its port by looking at the slave section on master:8080 (the web interface)
**GameClient**
Fancy game client that does the following steps automatically, but is really old so be warned: [clusterioClient](https://github.com/Danielv123/factorioClusterioClient)
## Running Clusterio
1. Download the same version of the mod as the slave is running from [the mod portal](https://mods.factorio.com/mods/Danielv123/clusterio) or [github](https://github.com/Danielv123/factorioClusterioMod
After following the installation instructions you can use the following
commands to run Clusterio.
2. Drop it into ./factorio/mods
3. Run factorio and connect to slave as a normal MP game. You will find the port number to connect to at http://[masterAddress]:8080
### Master Server
It's necessary to run the master server in order for anything to work.
Once you've completed the setup run the following command to start it
up:
node master
Or with pm2 (it's recommened to run it without pm2 first):
pm2 start master --name master
### Slaves
Slaves connect to the master server and are managed remotely from the
master server. In order for slaves to connect to the master server they
need an authentication token from the master server. This token is
written to secret-api-token.txt on the master server the first time it
is started.
To set up the configuration for a new local slave run the following.
node client create-config --name "Local" --token "<token>"
This will write a new `config-slave.json` file in the current directory
(you can change the location with the `--config` option) with the name
and token provided. If you are connecting to a remote master server you
will also need to pass the `--url <url to master server>` option.
You can edit the config of a slave with the `node client edit-config`
command. Use `node client edit-config --help` for more information.
Once the config is set up run the slave with
node client start
### Instances
Instances are created, managed and started from the master server. For
now the only interface available is the `clusterctl` command line tool
included in Clusterio. You can run this tool from any slave, or the
master server without having to set up a config, if you want to manage
the cluster from somewhere else see `node clusterctl create-config`.
The basic operations to start a new instance is the following
node clusterctl create-instance --slave "Local" --name "My instance"
node clusterctl create-save --instance "My instance"
node clusterctl start-instance --instance "My instance"
The first line creates the files and database entries for a new
instance. The second line creates a new savegame for the instance. You
could also upload your own save to the instance directory instead. And
finally the third line starts the instance.
There are many more commands available with clusterctl. See
`node clusterctl --help` for more information.
## Plugins
Here are the known Clusterio plugins in the wild:
@@ -257,7 +282,7 @@ Here are the known Clusterio plugins in the wild:
## Common problems
### Cannot find module: `/../../config`
### Cannot find module: `./config`
Copy your config.json.dist to config.json and configure it.
@@ -288,32 +313,3 @@ TLDR: the tested fix is:
### Other fixes for other potential problems:
Sometimes the install fails. Try `node ./lib/npmPostinstall` to complete it.
## Cheatsheet
**To create a new instance/start it**
node client start [instanceName]
**Other instance management tools:**
```
node client delete [instanceName]
node client list
```
**To update clusterio to the latest version:**
1. Download the latest zip version of factorio for your platform manually from factorio.com. Place it in the project root folder and call it "factorio" (folder name is specified in config.json)
2. Grab the latest version of the repo
```
git pull
npm install --only=production
```
3. Download the latest version of the factorioClusterioMod from its github repo
```
node client manage shared mods add clusterio
```
client.js
+493 -419
View File
@@ -5,18 +5,19 @@ const child_process = require('child_process');
const path = require('path');
const request = require("request");
const deepmerge = require("deepmerge");
const getMac = require('getmac').getMac;
const ioClient = require("socket.io-client");
const asTable = require("as-table").configure({delimiter: ' | '});
const util = require("util");
const yargs = require("yargs");
const version = require("./package").version;
// internal libraries
const objectOps = require("lib/objectOps");
const fileOps = require("lib/fileOps");
const pluginManager = require("lib/manager/pluginManager");
const modManager = require("lib/manager/modManager");
const hashFile = require('lib/hash').hashFile;
const factorio = require("lib/factorio");
const schema = require("lib/schema");
const link = require("lib/link");
// Uhm...
var global = {};
@@ -25,9 +26,200 @@ var global = {};
* Keeps track of the runtime parameters of an instance
*/
class Instance {
constructor(dir, name) {
this._name = name;
constructor(dir, factorioDir, instanceConfig) {
this._dir = dir;
// This is expected to change with the config system rewrite
this.config = {
id: instanceConfig.id,
name: instanceConfig.name,
gamePort: instanceConfig.factorioPort,
rconPort: instanceConfig.clientPort,
rconPassword: instanceConfig.clientPassword,
}
let serverOptions = {
gamePort: this.config.gamePort,
rconPort: this.config.rconPort,
rconPassword: this.config.rconPassword,
};
this.server = new factorio.FactorioServer(
path.join(factorioDir, "data"), this._dir, serverOptions
);
}
async init() {
await this.server.init();
}
static async create(id, instanceDir, factorioDir, options) {
let instanceConfig = {
id,
name: options.name,
factorioPort: null,
clientPort: null,
clientPassword: null,
}
let instance = new this(instanceDir, factorioDir, instanceConfig);
await instance.init();
console.log(`Creating ${instance.path()}`);
await fs.ensureDir(instance.path());
await fs.ensureDir(instance.path("script-output"));
await symlinkMods(instance, "sharedMods", console);
console.log("Clusterio | Created instance with settings:")
console.log(instanceConfig);
// save instance config
await fs.outputFile(instance.path("config.json"), JSON.stringify(instanceConfig, null, 4));
let serverSettings = await instance.server.exampleSettings();
let gameName = "Clusterio instance: " + instance.name;
if (options.username) {
gameName = options.username + "'s clusterio " + instance.name;
}
let overrides = {
"name": gameName,
"description": options.description,
"tags": ["clusterio"],
"visibility": options.visibility,
"username": options.username,
"token": options.token,
"game_password": options.game_password,
"require_user_verification": options.verify_user_identity,
"allow_commands": options.allow_commands,
"auto_pause": options.auto_pause,
};
for (let [key, value] of Object.entries(overrides)) {
if (!Object.hasOwnProperty.call(serverSettings, key)) {
throw Error(`Expected server settings to have a ${key} property`);
}
serverSettings[key] = value;
}
await fs.writeFile(instance.path("server-settings.json"), JSON.stringify(serverSettings, null, 4));
return instance;
}
async createSave() {
console.log("Creating save .....");
// XXX Move to constructor and send to master
this.server.on('output', function(output) {
console.log("Fact: " + output.message);
});
await this.server.create("world");
console.log("Clusterio | Successfully created save");
}
async start(slaveConfig, socket) {
console.log("Deleting .tmp.zip files");
let savefiles = fs.readdirSync(this.path("saves"));
for(let i = 0; i < savefiles.length; i++){
if(savefiles[i].substr(savefiles[i].length - 8, 8) == ".tmp.zip") {
fs.unlinkSync(this.path("saves", savefiles[i]));
}
}
console.log("Clusterio | Rotating old logs...");
// clean old log file to avoid crash
try{
let logPath = this.path("factorio-current.log");
let stat = await fs.stat(logPath);
console.log(stat)
console.log(stat.isFile())
if(stat.isFile()){
let logFilename = `factorio-${Math.floor(Date.parse(stat.mtime)/1000)}.log`;
await fs.rename(logPath, this.path(logFilename));
console.log(`Log rotated as ${logFilename}`);
}
}catch(e){}
await symlinkMods(this, "sharedMods", console);
// Spawn factorio server
let latestSave = await fileOps.getNewestFile(this.path("saves"));
if (latestSave === null) {
throw new Error(
"Your savefile seems to be missing. This might because you created an\n"+
"instance without having factorio installed and configured properly.\n"+
"Try installing factorio and adding your savefile to\n"+
"instances/[instancename]/saves/"
);
}
// Patch save with lua modules from plugins
console.log("Clusterio | Patching save");
// For now it's assumed that all files in the lua folder of a plugin is
// to be patched in under the name of the plugin and loaded for all
// plugins that are not disabled. This will most likely change in the
// future when the plugin refactor is done.
let modules = [];
for (let pluginName of await fs.readdir("sharedPlugins")) {
let pluginDir = path.join("sharedPlugins", pluginName);
if (await fs.pathExists(path.join(pluginDir, "DISABLED"))) {
continue;
}
if (!await fs.pathExists(path.join(pluginDir, "lua"))) {
continue;
}
let module = {
"name": pluginName,
"files": [],
};
for (let fileName of await fs.readdir(path.join(pluginDir, "lua"))) {
module["files"].push({
path: pluginName+"/"+fileName,
content: await fs.readFile(path.join(pluginDir, "lua", fileName)),
load: true,
});
}
modules.push(module);
}
await factorio.patch(this.path("saves", latestSave), modules);
this.server.on('output', (output) => {
console.log("Fact: " + output.message);
});
this.server.on('rcon-ready', () => {
console.log("Clusterio | RCON connection established");
// Temporary measure for backwards compatibility
let compatConfig = {
id: this.config.id,
unique: this.config.id,
name: this.config.name,
// FactorioServer.init may have generated a random port or password
// if they were null.
factorioPort: this.server.gamePort,
clientPort: this.server.rconPort,
clientPassword: this.server.rconPassword,
}
instanceManagement(slaveConfig, this, compatConfig, this.server, socket); // XXX async function
});
await this.server.start(latestSave);
}
/**
* Stop the instance
*/
async stop() {
// XXX this needs more thought to it
if (this.server._state === "running") {
await this.server.stop();
}
}
/**
@@ -36,7 +228,7 @@ class Instance {
* This should not be used for filesystem paths. See .path() for that.
*/
get name() {
return this._name;
return this.config.name;
}
/**
@@ -52,6 +244,169 @@ class Instance {
}
}
/**
* Handles running the slave
*
* Connects to the master server over the socket.io connection and manages
* intsances.
*/
class Slave extends link.Client {
// I don't like God classes, but the alternative of putting all this state
// into global variables is not much better.
constructor(slaveConfig) {
super('slave', slaveConfig.masterURL, slaveConfig.masterAuthToken);
link.attachAllMessages(this);
this.config = {
id: slaveConfig.id,
name: slaveConfig.name,
instancesDir: slaveConfig.instanceDirectory,
factorioDir: slaveConfig.factorioDirectory,
masterUrl: slaveConfig.masterURL,
masterToken: slaveConfig.masterAuthToken,
publicAddress: slaveConfig.publicIP,
}
}
register() {
console.log("SOCKET | registering slave");
this.send('register_slave', {
agent: 'Clusterio Slave',
version,
id: this.config.id,
name: this.config.name,
});
}
async _findNewInstanceDir(name) {
try {
checkFilename(name)
} catch (err) {
throw new Error(`Instance name ${err.message}`);
}
// For now add dashes until an unused directory name is found
let dir = path.join(this.config.instancesDir, name);
while (await fs.pathExists(dir)) {
dir += '-';
}
return dir;
}
/**
* Looks up the instance for an instance request handler
*
* Called by the handler for InstanceRequest.
*/
async forwardInstanceRequest(message, handler) {
let instance = this.instances.get(message.data.instance_id);
if (!instance) {
throw new errors.RequestError(`Instance with ID ${instanceId} does not exist`);
}
return await handler.call(this, instance, message);
}
async createInstanceRequestHandler(message) {
let { id, options } = message.data;
if (this.instances.has(id)) {
throw new Error(`Instance with ID ${id} already exists`);
}
let instanceDir = await this._findNewInstanceDir(options.name);
// XXX: race condition on multiple simultanious calls
let instance = await Instance.create(id, instanceDir, this.config.factorioDir, options);
this.instances.set(id, instance);
this.updateInstances();
}
async createSaveRequestHandler(instance, message) {
await instance.createSave();
}
async startInstanceRequestHandler(instance, message) {
await instance.start(this.config, this.socket);
}
async stopInstanceRequestHandler(instance, message) {
await instance.stop();
}
async deleteInstanceRequestHandler(instance, message) {
await instance.stop();
this.instances.delete(instance.config.id);
fs.remove(instance.path());
this.updateInstances();
}
async sendRconRequestHandler(instance, message) {
let result = await instance.server.sendRcon(message.data.command);
return { result };
}
async findInstances() {
let instances = new Map();
for (let entry of await fs.readdir(this.config.instancesDir, { withFileTypes: true })) {
if (entry.isDirectory()) {
let instanceConfig;
let configPath = path.join(this.config.instancesDir, entry.name, "config.json");
try {
instanceConfig = JSON.parse(await fs.readFile(configPath));
} catch (err) {
if (err.code === "ENOENT") {
continue; // Ignore folders without config.json
}
console.error(`Error occured while parsing ${configPath}: ${err}`);
}
// XXX should probably validate the entire config with a JSON Schema.
if (typeof instanceConfig.id !== "number" || isNaN(instanceConfig.id)) {
console.error(`${configPath} is missing id`);
continue;
}
if (typeof instanceConfig.name !== "string") {
console.error(`${configPath} is missing name`);
continue;
}
let instancePath = path.join(this.config.instancesDir, entry.name);
console.log(`found instance ${instanceConfig.name} in ${instancePath}`);
let instance = new Instance(instancePath, this.config.factorioDir, instanceConfig);
await instance.init();
instances.set(instanceConfig.id, instance);
}
}
return instances;
}
updateInstances() {
let list = [];
for (let instance of this.instances.values()) {
list.push({
id: instance.config.id,
name: instance.config.name,
});
}
link.events.updateInstances.send(this, { instances: list });
}
async start() {
this.instances = await this.findInstances();
await this.connect();
this.updateInstances();
}
async stop() {
for (let instance of this.instances.values()) {
await instance.stop();
}
}
}
function checkFilename(name) {
// All of these are bad in Windows only, except for /, . and ..
// See: https://docs.microsoft.com/en-us/windows/win32/fileio/naming-a-file
@@ -92,121 +447,6 @@ function checkFilename(name) {
}
}
function needleOptionsWithTokenAuthHeader(config) {
return {
compressed: true,
headers: {
'x-access-token': config.masterAuthToken
},
};
}
function printUsage() {
console.error("Usage: ");
console.error("node client start [instance name]");
console.error("node client list");
console.error("node client delete [instance]");
console.error("To download the latest version of the Clusterio lua mod, do");
console.error("node client manage shared mods download clusterio");
console.error("To download a clusterio plugin, do");
console.error("node client manage shared plugins install https://github.com/Danielv123/playerManager");
console.error("For more management options, do");
console.error("node client manage");
}
async function listInstances(config) {
let instanceNames = fileOps.getDirectoriesSync(config.instanceDirectory);
let instances = [];
for (instance of instanceNames) {
let cfg = path.resolve(config.instanceDirectory, instance, 'config');
let port = require(cfg).factorioPort;
instances.push({
"Name": instance,
"Port": port.toString(),
});
}
console.log(asTable(instances));
}
async function manage(config, instance) {
// console.log("Launching mod manager");
//const fullUsage = 'node client manage [instance, "shared"] ["mods", "config"] ...';
function usage(instance, tool, action){
if(tool && tool == "mods"){
console.log('node client manage '+instance.name+' '+tool+' ["list", "search", "add", "remove"]');
} else if(tool && tool == "plugins") {
console.log(`node client manage ${instance.name} ${tool} ["list", "add", "remove"]`);
} else {
console.log('node client manage '+(instance && instance.name || '[instance, "shared"]') +' '+ (tool || '["mods", "plugins"]') + ' ...');
}
}
const tool = process.argv[4] || "";
const action = process.argv[5] || "";
if (instance !== undefined) {
if(tool == "mods"){
(async function(){try{
// allow managing mods
if(action == "list"){
console.log(await modManager.listMods(config, instance));
} else if(action == "search"){
console.log(await modManager.findMods(config, process.argv[6]));
} else if(action == "add" || action == "download"){
await modManager.addMod(config, process.argv[6], instance);
} else if(action == "remove" || action == "rm" || action == "delete"){
await modManager.removeMod(config, process.argv[6], instance);
} else {
usage(instance, tool);
}
process.exit(0);
}catch(e){
console.log("Got error from modManager:")
console.log(e);
}})();
} else if(tool == "plugins"){
(async function(){try{
if(action == "list"){
await pluginManager.listPlugins();
} else if(action == "add" || action == "install" || action == "download"){
let status = await pluginManager.addPlugin(process.argv[6]);
} else if(action == "remove" || action == "uninstall" || action == "delete"){
let status = await pluginManager.removePlugin(process.argv[6]);
if(status && status.msg) console.log(status.msg);
} else if(action == "enable"){
let status = await pluginManager.enablePlugin(process.argv[6]);
if(status && status.msg) console.log(status.msg);
} else if(action == "disable"){
let status = await pluginManager.disablePlugin(process.argv[6]);
if(status && status.msg) console.log(status.msg);
}
process.exit(0);
}catch(e){
console.error(e);
process.exit(1);
}})();
} else {
usage(instance);
}
} else {
console.log('Usage:');
usage(instance);
}
}
async function deleteInstance(instance) {
if (instance === undefined) {
console.error("Usage: node client delete [instance]");
process.exit(1);
} else if (fs.existsSync(instance.path())) {
fileOps.deleteFolderRecursiveSync(instance.path()); // TODO: Check if this can cause i-craft users to format their server by using wrong paths
console.log("Deleted instance " + instance.name);
process.exit(0);
} else {
console.error("Instance not found: " + instance.name);
process.exit(0);
}
}
/**
* Create and update symlinks for shared mods in an instance
*
@@ -270,228 +510,103 @@ async function symlinkMods(instance, sharedMods, logger) {
}
}
async function createInstance(config, instance) {
console.log(`Creating ${instance.path()}`);
await fs.ensureDir(instance.path());
await fs.ensureDir(instance.path("script-output"));
await symlinkMods(instance, "sharedMods", console);
let instconf = {
"id": Math.random() * 2 ** 31 | 0,
"factorioPort": null,
"clientPort": null,
"__comment_clientPassword": "This is the rcon password. Will be randomly generated if null.",
"clientPassword": null,
"info": {}
}
console.log("Clusterio | Created instance with settings:")
console.log(instconf);
// create instance config
fs.writeFileSync(instance.path("config.json"), JSON.stringify(instconf, null, 4));
let server = new factorio.FactorioServer(path.join("factorio", "data"), instance.path(), {});
await server.init();
let serverSettings = await server.exampleSettings();
let name = "Clusterio instance: " + instance.name;
if (config.username) {
name = config.username + "'s clusterio " + instance.name;
}
let overrides = {
"name": name,
"description": config.description,
"tags": ["clusterio"],
"max_players": "20",
"visibility": config.visibility,
"username": config.username,
"token": config.token,
"game_password": config.game_password,
"require_user_verification": config.verify_user_identity,
"allow_commands": config.allow_commands,
"autosave_interval": 10,
"autosave_slots": 5,
"afk_autokick_interval": 0,
"auto_pause": config.auto_pause,
};
for (let [name, value] of Object.entries(overrides)) {
if (!Object.hasOwnProperty.call(serverSettings, name)) {
throw Error(`Expected server settings to have a ${name} property`);
}
serverSettings[name] = value;
}
await fs.writeFile(instance.path("server-settings.json"), JSON.stringify(serverSettings, null, 4));
console.log("Server settings: ", serverSettings);
console.log("Creating save .....");
server.on('output', function(output) {
console.log("Fact: " + output.message);
});
await server.create("world");
console.log("Clusterio | Successfully created instance");
}
async function startInstance(config, instance) {
var instanceconfig = JSON.parse(await fs.readFile(instance.path("config.json")));
if (typeof instanceconfig.id !== "number" || isNaN(instanceconfig.id)) {
throw new Error(`${instance.path("config.json")} is missing id`);
}
// Temporary measure for backwards compatibility
instanceconfig.unique = instanceconfig.id;
console.log("Deleting .tmp.zip files");
let savefiles = fs.readdirSync(instance.path("saves"));
for(i = 0; i < savefiles.length; i++){
if(savefiles[i].substr(savefiles[i].length - 8, 8) == ".tmp.zip") {
fs.unlinkSync(instance.path("saves", savefiles[i]));
}
}
console.log("Clusterio | Rotating old logs...");
// clean old log file to avoid crash
try{
let logPath = instance.path("factorio-current.log");
let stat = await fs.stat(logPath);
console.log(stat)
console.log(stat.isFile())
if(stat.isFile()){
let logFilename = `factorio-${Math.floor(Date.parse(stat.mtime)/1000)}.log`;
await fs.rename(logPath, instance.path(logFilename));
console.log(`Log rotated as ${logFilename}`);
}
}catch(e){}
await symlinkMods(instance, "sharedMods", console);
// Spawn factorio server
let latestSave = await fileOps.getNewestFile(instance.path("saves"));
if (latestSave === null) {
throw new Error(
"Your savefile seems to be missing. This might because you created an\n"+
"instance without having factorio installed and configured properly.\n"+
"Try installing factorio and adding your savefile to\n"+
"instances/[instancename]/saves/"
);
}
// Patch save with lua modules from plugins
console.log("Clusterio | Patching save");
// For now it's assumed that all files in the lua folder of a plugin is
// to be patched in under the name of the plugin and loaded for all
// plugins that are not disabled. This will most likely change in the
// future when the plugin refactor is done.
let modules = [];
for (let pluginName of await fs.readdir("sharedPlugins")) {
let pluginDir = path.join("sharedPlugins", pluginName);
if (await fs.pathExists(path.join(pluginDir, "DISABLED"))) {
continue;
}
if (!await fs.pathExists(path.join(pluginDir, "lua"))) {
continue;
}
let module = {
"name": pluginName,
"files": [],
};
for (let fileName of await fs.readdir(path.join(pluginDir, "lua"))) {
module["files"].push({
path: pluginName+"/"+fileName,
content: await fs.readFile(path.join(pluginDir, "lua", fileName)),
load: true,
});
}
modules.push(module);
}
await factorio.patch(instance.path("saves", latestSave), modules);
let options = {
gamePort: instanceconfig.factorioPort,
rconPort: instanceconfig.clientPort,
rconPassword: instanceconfig.clientPassword,
};
let server = new factorio.FactorioServer(path.join("factorio", "data"), instance.path(), options);
await server.init();
// FactorioServer.init may have generated a random port or password
// if they were null.
instanceconfig.factorioPort = server.gamePort
instanceconfig.clientPort = server.rconPort
instanceconfig.clientPassword = server.rconPassword
server.on('output', function(output) {
console.log("Fact: " + output.message);
});
server.on('rcon-ready', function() {
console.log("Clusterio | RCON connection established");
instanceManagement(config, instance, instanceconfig, server); // XXX async function
});
let secondSigint = false
process.on('SIGINT', () => {
if (secondSigint) {
console.log("Caught second interrupt, terminating immediately");
process.exit(1);
}
secondSigint = true;
console.log("Caught interrupt signal, shutting down");
server.stop().then(() => {
// There's currently no shutdown mechanism for instance plugins so
// they keep the event loop alive.
process.exit();
});
});
await server.start(latestSave);
}
async function startClient() {
// argument parsing
const args = require('minimist')(process.argv.slice(2));
const command = process.argv[2];
// add better stack traces on promise rejection
process.on('unhandledRejection', r => console.log(r));
console.log(`Requiring config from ${args.config || './config'}`);
const config = require(args.config || './config');
// argument parsing
const args = yargs
.scriptName("client")
.usage("$0 <command> [options]")
.option('config', {
nargs: 1,
describe: "slave config file to use",
default: 'config-slave.json',
type: 'string',
})
.command('create-config', "Create slave config", (yargs) => {
yargs.options({
'name': { describe: "Name of the slave", nargs: 1, type: 'string', demandOption: true },
'url': { describe: "Master URL", nargs: 1, type: 'string', default: "http://localhost:8080/" },
'token': { describe: "Master token", nargs: 1, type: 'string', demandOption: true },
'ip': { describe: "Public facing IP", nargs: 1, type: 'string', default: "localhost" },
'instances-dir': { describe: "Instances directory", nargs: 1, type: 'string', default: "instances" },
'factorio-dir': { describe: "Factorio directory", nargs: 1, type: 'string', default: "factorio" },
'id': {
describe: "Numeric id of the slave",
nargs: 1,
type: 'number',
default: Math.random() * 2**31 | 0,
defaultDescription: "random id",
},
});
})
.command('edit-config', "Edit slave config", (yargs) => {
yargs.options({
'name': { describe: "Set name of the slave", nargs: 1, type: 'string' },
'url': { describe: "Set master URL", nargs: 1, type: 'string' },
'token': { describe: "Set master token", nargs: 1, type: 'string' },
'ip': { describe: "Set public facing IP", nargs: 1, type: 'string' },
'instances-dir': { describe: "Set instances directory", nargs: 1, type: 'string' },
'factorio-dir': { describe: "Set Factorio directory", nargs: 1, type: 'string' },
'id': { describe: "Set id of the slave", nargs: 1, type: 'number' },
});
})
.command('show-config', "Show slave config")
.command('start', "Start slave")
.demandCommand(1, "You need to specify a command to run")
.strict()
.argv
;
await fs.ensureDir(config.instanceDirectory);
let command = args._[0];
if (command === "create-config") {
await fs.outputFile(args.config, JSON.stringify({
name: args.name,
masterURL: args.url,
masterAuthToken: args.token,
publicIP: args.ip,
instanceDirectory: args.instancesDir,
factorioDirectory: args.factorioDir,
id: args.id,
}, null, 4), { flag: 'wx' });
return;
} else if (command == "edit-config") {
let slaveConfig = JSON.parse(await fs.readFile(args.config));
if ('name' in args) slaveConfig.name = args.name;
if ('url' in args) slaveConfig.masterURL = args.url;
if ('token' in args) slaveConfig.masterAuthToken = args.token;
if ('ip' in args) slaveConfig.publicIP = args.ip;
if ('instancesDir' in args) slaveConfig.instanceDirectory = args.instancesDir;
if ('factorioDir' in args) slaveConfig.factorioDirectory = args.factorioDir;
if ('id' in args) slaveConfig.id = args.id;
await fs.outputFile(args.config, JSON.stringify(slaveConfig, null, 4));
return;
} else if (command == "show-config") {
let slaveConfig = JSON.parse(await fs.readFile(args.config));
console.log(slaveConfig);
return;
}
// If we get here the command was start
// handle commandline parameters
console.log(`Loading ${args.config}`);
const slaveConfig = JSON.parse(await fs.readFile(args.config));
await fs.ensureDir(slaveConfig.instanceDirectory);
await fs.ensureDir("sharedPlugins");
await fs.ensureDir("sharedMods");
let instance;
if (process.argv[3] !== undefined) {
let name = process.argv[3];
try {
checkFilename(name);
} catch (err) {
throw new Error(`Instance name ${err.message}`);
}
let dir = path.join(config.instanceDirectory, name);
instance = new Instance(dir, name);
}
// Set the process title, shows up as the title of the CMD window on windows
// and as the process name in ps/top on linux.
if (instance) {
process.title = "clusterioClient "+instance.name;
}
process.title = "clusterioClient";
// make sure we have the master access token (can't write to master without it since clusterio 2.0)
if(!config.masterAuthToken || typeof config.masterAuthToken !== "string"){
// make sure we have the master access token
if(!slaveConfig.masterAuthToken || typeof slaveConfig.masterAuthToken !== "string"){
console.error("ERROR invalid config!");
console.error(
"Master server now needs an access token for write operations. As clusterio\n"+
@@ -503,52 +618,64 @@ async function startClient() {
return;
}
// handle commandline parameters
if (!command || command == "help" || command == "--help") {
printUsage();
process.exit(1);
} else if (command == "list") {
await listInstances(config);
process.exit(0);
// make sure url ends with /
if (!slaveConfig.masterURL.endsWith("/")) {
console.error("ERROR invalid config!");
console.error("masterURL (set with --url) must end with '/'");
process.exitCode = 1;
return;
}
let slave = new Slave(slaveConfig);
// Handle interrupts
let secondSigint = false
process.on('SIGINT', () => {
if (secondSigint) {
console.log("Caught second interrupt, terminating immediately");
process.exit(1);
}
secondSigint = true;
console.log("Caught interrupt signal, shutting down");
slave.stop().then(() => {
// There's currently no shutdown mechanism for instance plugins so
// they keep the event loop alive.
process.exit();
});
});
await slave.start();
/*
} else if (command == "manage"){
await manage(config, instance);
// process.exit(0);
} else if (command == "delete") {
await deleteInstance(instance);
} else if (command == "start" && instance === undefined) {
console.error("ERROR: No instanceName provided!");
console.error("Usage: node client start [instanceName]");
process.exit(0);
} else if (command == "start" && !fs.existsSync(instance.path())) {
await createInstance(config, instance);
} else if (command == "start" && fs.existsSync(instance.path())) {
await startInstance(config, instance);
} else {
console.error("Invalid arguments, quitting.");
process.exit(1);
}
*/
}
// ensure instancemanagement only ever runs once
var _instanceInitialized;
async function instanceManagement(config, instance, instanceconfig, server) {
async function instanceManagement(slaveConfig, instance, instanceconfig, server, socket) {
if (_instanceInitialized) return;
_instanceInitialized = true;
console.log("Started instanceManagement();");
let compatUrl = slaveConfig.masterUrl;
if (compatUrl.endsWith("/")) {
compatUrl = compatUrl.slice(0, -1);
}
let compatConfig = {
name: slaveConfig.name,
masterURL: compatUrl,
masterAuthToken: slaveConfig.masterToken,
publicIP: slaveConfig.publicAddress,
instanceDirectory: slaveConfig.instancesDir,
factorioDirectory: slaveConfig.factorioDir,
}
/* Open websocket connection to master */
var socket = ioClient(config.masterURL+"?token="+config.masterAuthToken);
socket.on("error", err => {
console.log("SOCKET | Error: ", err);
});
socket.on("hello", data => {
console.log("SOCKET | registering slave!");
socket.emit("registerSlave", {
instanceID: instanceconfig.unique,
});
});
setInterval(B=> socket.emit("heartbeat"), 10000);
console.log("Started instanceManagement();");
// load plugins and execute onLoad event
let pluginsToLoad = await pluginManager.getPlugins();
@@ -556,7 +683,7 @@ async function instanceManagement(config, instance, instanceconfig, server) {
for(let i = 0; i < pluginsToLoad.length; i++){
let pluginLoadStarted = Date.now();
let combinedConfig = deepmerge(instanceconfig,config,{clone:true});
let combinedConfig = deepmerge(instanceconfig,compatConfig,{clone:true});
combinedConfig.instanceName = instance.name;
let pluginConfig = pluginsToLoad[i];
@@ -651,59 +778,6 @@ async function instanceManagement(config, instance, instanceconfig, server) {
// this plugin doesn't have a client portion. Maybe it runs on the master only?
}
}
// world IDs ------------------------------------------------------------------
hashMods(instance, function(modHashes){
setInterval(getID, 10000);
getID();
function getID() {
server.sendRcon("/silent-command rcon.print(#game.connected_players)").then((playerCount) => {
var payload = {
time: Date.now(),
rconPort: instanceconfig.clientPort,
rconPassword: instanceconfig.clientPassword,
serverPort: instanceconfig.factorioPort,
unique: instanceconfig.unique,
publicIP: config.publicIP, // IP of the server should be global for all instances, so we pull that straight from the config
mods:modHashes,
instanceName: instance.name,
info: instanceconfig.info,
};
if(playerCount){
payload.playerCount = playerCount.replace(/(\r\n\t|\n|\r\t)/gm, "");
} else {
payload.playerCount = 0;
}
function callback(err, mac) {
if (err) {
mac = "unknown";
console.log("##### getMac crashed, but we don't really give a shit because we are probably closing down #####");
}
global.mac = mac;
payload.mac = mac;
// console.log("Registered our presence with master "+config.masterURL+" at " + payload.time);
needle.post(config.masterURL + '/api/getID', payload, needleOptionsWithTokenAuthHeader(config), function (err, response, body) {
if (err && err.code != "ECONNRESET"){
console.error("We got problems, something went wrong when contacting master "+config.masterURL+" at " + payload.time);
console.error(err);
} else 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.
if(response.body !== "ok") {
console.log("Got no \"ok\" while registering our precense with master "+config.masterURL+" at " + payload.time);
console.log(response.body);
}
}
});
}
if(global.mac){
callback(undefined, global.mac);
} else {
getMac(callback);
}
});
}
});
} // END OF INSTANCE START ---------------------------------------------------------------------
// string, function
clusterctl.js
+382
View File
@@ -0,0 +1,382 @@
const io = require("socket.io-client");
const jwt = require("jsonwebtoken");
const fs = require("fs-extra");
const yargs = require("yargs");
const version = require("./package").version;
const asTable = require("as-table").configure({delimiter: ' | '});
const link = require("lib/link");
const errors = require("lib/errors");
/**
* Represens a command that can be runned by clusterctl
*/
class Command {
constructor({ definition, handler }) {
this.name = definition[0];
this._definition = definition;
this._handler = handler;
}
register(yargs) {
yargs.command(...this._definition);
}
async run(args, control) {
await this._handler.call(this, args, control);
}
}
/**
* Resolve a string to an instance ID
*
* Resolves a string with either an instance name or an id into an integer
* with the instance ID.
*
* @param link - link to master server to query instance on.
* @param instance - string with name or id of instance.
* @returns {number} instance ID.
*/
async function resolveInstance(control, instanceName) {
let instanceId;
if (/^-?\d+$/.test(instanceName)) {
instanceId = parseInt(instanceName, 10);
} else {
let response = await link.requests.listInstances.send(control);
for (let instance of response.list) {
if (instance.name === instanceName) {
instanceId = instance.id;
break;
}
}
if (instanceId === undefined) {
throw new errors.CommandError(`No instance named ${instanceName}`);
}
}
return instanceId;
}
let commands = [];
commands.push(new Command({
definition: ["list-slaves", "List slaves connected to the master"],
handler: async function(args, control) {
let response = await link.requests.listSlaves.send(control);
console.log(asTable(response.list));
},
}));
commands.push(new Command({
definition: ["list-instances", "List instances known to the master"],
handler: async function(args, control) {
let response = await link.requests.listInstances.send(control);
console.log(asTable(response.list));
}
}));
commands.push(new Command({
definition: ['create-instance', "Create an instance", (yargs) => {
yargs.options({
'name': { describe: "Name of the instance", nargs: 1, type: 'string', demandOption: true },
'slave': { describe: "Slave to create on", nargs: 1, type: 'string', demandOption: true },
});
}],
handler: async function(args, control) {
let slaveId;
if (/^-?\d+$/.test(args.slave)) {
slaveId = parseInt(args.slave, 10);
} else {
let response = await link.requests.listSlaves.send(control);
for (let slave of response.list) {
if (slave.name === args.slave) {
slaveId = slave.id;
break;
}
}
if (slaveId === undefined) {
throw new errors.CommandError(`No slave named ${args.slave}`);
}
}
let response = await link.requests.createInstanceCommand.send(control, {
name: args.name,
slave_id: slaveId,
});
console.log(response);
},
}));
commands.push(new Command({
definition: ['create-save', "Create a new save on an instance", (yargs) => {
yargs.options({
'instance': { describe: "Instance to create on", nargs: 1, type: 'string', demandOption: true },
});
}],
handler: async function(args, control) {
let response = await link.requests.createSave.send(control, {
instance_id: await resolveInstance(control, args.instance),
});
console.log(response);
},
}));
commands.push(new Command({
definition: ['start-instance', "Start instance", (yargs) => {
yargs.options({
'instance': { describe: "Instance to start", nargs: 1, type: 'string', demandOption: true },
});
}],
handler: async function(args, control) {
let response = await link.requests.startInstance.send(control, {
instance_id: await resolveInstance(control, args.instance),
});
console.log(response);
},
}));
commands.push(new Command({
definition: ['stop-instance', "Stop instance", (yargs) => {
yargs.options({
'instance': { describe: "Instance to stop", nargs: 1, type: 'string', demandOption: true },
});
}],
handler: async function(args, control) {
let response = await link.requests.stopInstance.send(control, {
instance_id: await resolveInstance(control, args.instance),
});
console.log(response);
},
}));
commands.push(new Command({
definition: ['delete-instance', "Delete instance", (yargs) => {
yargs.options({
'instance': { describe: "Instance to delete", nargs: 1, type: 'string', demandOption: true },
});
}],
handler: async function(args, control) {
let response = await link.requests.deleteInstance.send(control, {
instance_id: await resolveInstance(control, args.instance),
});
console.log(response);
},
}));
commands.push(new Command({
definition: ['send-rcon', "Send RCON command", (yargs) => {
yargs.options({
'instance': { describe: "Instance to sent to", nargs: 1, type: 'string', demandOption: true },
'command': { describe: "command to send", nargs: 1, type: 'string', demandOption: true },
});
}],
handler: async function(args, control) {
let response = await link.requests.sendRcon.send(control, {
instance_id: await resolveInstance(control, args.instance),
command: args.command
})
// Factorio includes a newline in it's response output.
process.stdout.write(response.result);
},
}));
// Convert to mapping from name to command instance
commands = new Map([...commands.map(command => [command.name, command])]);
/**
* Handles running the control
*
* Connects to the master server over the socket.io connection and sends
* commands to it.
*/
class Control extends link.Client {
// I don't like God classes, but the alternative of putting all this state
// into global variables is not much better.
constructor(controlConfig) {
super('control', controlConfig.url, controlConfig.token);
link.attachAllMessages(this);
}
register() {
console.log("SOCKET | registering control");
this.send('register_control', {
agent: 'clusterctl',
version: version,
});
}
}
async function findCredentials() {
// Try looking in the default master config location
try {
let masterConfig = JSON.parse(await fs.readFile("config.json"));
let token = jwt.sign({ id: "api" }, masterConfig.masterAuthSecret, { expiresIn: 600 });
let url;
if (masterConfig.sslPort) {
url = `https://localhost:${masterConfig.sslPort}`;
} else {
url = `http://localhost:${masterConfig.masterPort}`;
}
return { url, token };
} catch (err) {
if (err.code !== "ENOENT") {
throw err;
}
}
// Try looking in the default slave config location
// In the future it's likely that tokens will be assigned more granular
// authorizations, meaning that you cannot use a slave token to control
// the cluster.
try {
let slaveConfig = JSON.parse(await fs.readFile("config-slave.json"));
return {
url: slaveConfig.masterURL,
token: slaveConfig.masterAuthToken,
}
} catch (err) {
if (err.code !== "ENOENT") {
throw err;
}
}
throw new errors.StartupError("Couldn't find credentials to connect to the master server with");
}
async function startControl() {
yargs
.scriptName("clusterctl")
.usage("$0 <command> [options]")
.option('config', {
nargs: 1,
describe: "config file to get credentails from",
default: "config-control.json",
defaultDescription: "auto",
type: 'string',
})
.command('create-config', "Create control config", (yargs) => {
yargs.options({
'url': { describe: "Master URL", nargs: 1, type: 'string', default: "http://localhost:8080/" },
'token': { describe: "Master token", nargs: 1, type: 'string', demandOption: true },
});
})
.command('edit-config', "Edit control config", (yargs) => {
yargs.options({
'url': { describe: "Set master URL", nargs: 1, type: 'string' },
'token': { describe: "Set master token", nargs: 1, type: 'string' },
});
})
.command('show-config', "Show control config")
;
for (let command of commands.values()) {
command.register(yargs);
}
const args = yargs.demandCommand(1, "You need to specify a command to run")
.strict()
.argv
;
let commandName = args._[0];
if (commandName === "create-config") {
await fs.outputFile(args.config, JSON.stringify({
url: args.url,
token: args.token,
}, null, 4), { flag: 'wx' });
return;
} else if (commandName === "edit-config") {
let controlConfig = JSON.parse(await fs.readFile(args.config));
if ('url' in args) controlConfig.url = args.url;
if ('token' in args) controlConfig.token = args.token;
await fs.outputFile(args.config, JSON.stringify(controlConfig, null, 4));
return;
} else if (commandName === "show-config") {
let controlConfig = JSON.parse(await fs.readFile(args.config));
console.log(controlConfig);
return;
}
// The remaining commands require connecting to the master server.
let controlConfig;
try {
controlConfig = JSON.parse(await fs.readFile(args.config));
} catch (err) {
if (err.code === "ENOENT") {
controlConfig = await findCredentials();
} else {
throw err;
}
}
let control = new Control(controlConfig);
await control.connect();
if (commands.has(commandName)) {
command = commands.get(commandName);
try {
await command.run(args, control)
} catch (err) {
if (err instanceof errors.CommandError) {
console.error(`Error running command: ${err.message}`);
process.exitCode = 1;
} else if (err instanceof errors.RequestError) {
console.error(`Error sending request: ${err.message}`);
process.exitCode = 1;
} else {
control.close("error");
throw err;
}
}
control.close("quit");
return; // ??
}
//XXX control.close("done");
}
if (module === require.main) {
console.warn(`
+==========================================================+
I WARNING: This is the development branch for the 2.0 I
I version of clusterio. Expect things to break. I
+==========================================================+
`
);
startControl().catch(err => {
if (!(err instanceof errors.StartupError)) {
console.error(`
+----------------------------------------------------------------+
| Unexpected error occured while starting control, please report |
| it to https://github.com/clusterio/factorioClusterio/issues |
+----------------------------------------------------------------+`
);
} else {
console.error(`
+-------------------------------+
| Unable to to start clusterctl |
+-------------------------------+`
);
}
console.error(err);
process.exitCode = 1;
});
}
config.json.dist
-10
View File
@@ -1,15 +1,8 @@
{
"__comment": "Where factorio folder. Should be in the same folder as this config file.",
"factorioDirectory": "./factorio",
"instanceDirectory": "./instances",
"databaseDirectory": "./database",
"__comment1": "IP players should connect to to join your servers, should be portforwarded and stuff",
"publicIP": "localhost",
"__comment2": "URL the master web server runs at, can be ignored on the master system. Must not end with a slash",
"masterURL": "http://localhost:8080",
"__commentMasterPort": "HTTP port for the master to listen to",
"masterPort": 8080,
@@ -18,9 +11,6 @@
"sslCert": "database/certificates/cert.crt",
"sslPrivKey": "database/certificates/cert.key",
"__commentMasterAuthToken": "Authentication token to connect to the master server. Find it in secret-api-token.txt on the master.",
"masterAuthToken": "",
"__commentAuthSecret": "This is a long random string used for generating authentication tokens. Do not share. Regenerates when secret-api-token.txt is deleted.",
"masterAuthSecret": "",
docs/socket.md
+115
View File
@@ -0,0 +1,115 @@
Socket.io protocol
==================
<sub>This document describes the protocol used between the master and
slaves, and unless you're developing Clusterio it's probably not going
to be very useful for you.</sub>
The protocol is built on the [socket.io](https://socket.io/) library
which lets mulitple streams be multiplexed in many different ways over
the same connection and abstracts away reconnection logic.
Unfortunately I've been unable to figure what the message delivery
guarantees of the library is, so whether or not messages can be lost is
still an unknown, I assume that messages are delivered as long as a
connection is maintained or manages to reconnect.
While there are many mechanisms to multiplex streams, the core
communication between the master server and slaves is done entirely by
sending single POJO payloads over the `message` event. There's a
shortcut for sending these via the `.send` method of the socket.io
sockets.
These messages contain the following three properties:
- seq - integer - The sequence number of the message
- type - string - The type of message
- data - object - Message data
Handshake
---------
Upon connecting to the socket.io endpoint the server will respond with a
`hello` message containing the master server version as the data. The
client is then expected to send a `register_slave` or `register_control`
message back depending on what kind of client it is. If the register
operation is successful the master will reply with a `ready` message
with an empty data payload.
**Master `hello` data**
- version - string - The version of the master, e.g. "2.0.0".
**Slave `register_slave` response data**
- agent - string - Human readable identifying the software connecting.
- version - string - The protocol version of the slave, e.g. "2.0.0".
- name - string - Name of the slave.
- id - integer - ID of the slave.
**Control `register_control repsonse data**
- agent - string - Human readable identifying the software connecting.
- version - string - The protocol version of the control, e.g. "2.0.0".
**Example handshake**
```js
// Server hello
({
type: 'hello',
data: {
version: "2.0.0",
}
})
// Client response
({
type: 'register_slave',
data: {
agent: 'Clusterio Slave',
version: "2.0.0",
name: "Foo",
id: 123,
}
})
// Server repsonse
({
type: 'ready',
data: {}
})
```
Requests
--------
Requests are messages which expects a response in return. They function
a lot like HTTP requests except that both parties of a connection can
inniate a request that is expected to be responded to. By convention
message types ending in `*_request` is a request, and is expected to be
replied to with a corresponding `*_response` message. If an error
occured while processing the request an error response containing an
error property in the data payload is sent instead.
**`*_response` data**
- seq - integer - The seq number of the request message.
- additional properties may be present depending on the type of request.
**`*_resopnse` error data**
- seq - integer - The seq number of the request message.
- error - string - Human readable text message describing the error.
See [lib/link/requests.js](lib/link/requests.js) for the recognized
requests and their implementation.
Closing the connection
----------------------
The party that innitates the shut down of the connection should send a
close message containing the reason for closing before disconnecting.
**`close` data**
- reason - string - Reason for closing the connection
lib/clusterTools/getInstanceName.js
-40
View File
@@ -1,40 +0,0 @@
const needle = require("needle");
module.exports = function(config){
const instances = [];
if(config){
return function getInstanceName(instanceID){
return new Promise((resolve, reject) => {
let instance = instances[instanceID];
if(!instance){
needle.get(config.masterURL + '/api/slaves', { compressed: true }, (err, response) => {
if(err || response.statusCode != 200) {
console.log("Unable to get JSON master/api/slaves, master might be unaccessible");
} else if (response && response.body) {
if(Buffer.isBuffer(response.body)) {console.log(response.body.toString("utf-8")); reject(new Error())}
try {
for (let index in response.body)
instances[index] = response.body[index].instanceName;
} catch (e){
console.log(e);
return null;
}
instance = instances[instanceID]
if (!instance) instance = instanceID; //somehow the master doesn't know the instance
resolve(instance);
}
});
} else {
resolve(instance);
}
});
}
} else {
return async function thisAintGonnaWork(){
let err = new Error("getInstanceName from lib/clusterTools needs to be called with {config} to work");
console.log(err);
throw err;
}
}
}
lib/clusterTools/index.js
-16
View File
@@ -1,16 +0,0 @@
/**
Functions frequently needed from different places in the cluster. Ex:
* Getting an instance's displayname from their ID
*/
module.exports = ({config} = {}) => ({
/**
async function to get an instances displayname from their instanceID
arg1 is the instanceID
Promise resolves with a string, either the displayName or instanceID if not found.
*/
getInstanceName: require("./getInstanceName")(config),
})
lib/errors.js
+20
View File
@@ -0,0 +1,20 @@
/**
* Errors thrown in Clusterio
* @module
*/
class CommandError extends Error { }
class RequestError extends Error { }
// Signal for messages that fail validation
class InvalidMessage extends Error { }
// Errror class for known errors occuring during startup
class StartupError extends Error { }
module.exports = {
CommandError,
RequestError,
InvalidMessage,
StartupError,
}
lib/factorio/server.js
+1 -1
View File
@@ -612,7 +612,7 @@ class FactorioServer extends events.EventEmitter {
* @returns {string} response from server.
*/
async sendRcon(message) {
this._check(["init", "running"])
this._check(["running"])
if (!this._rconReady) {
await events.once(this, 'rcon-ready');
}
lib/getInstances.js
-17
View File
@@ -1,17 +0,0 @@
const fileOps = require("lib/fileOps");
// const config = require("./../config");
module.exports = function(config) {
return {
sync: function(){
let instances;
try {
instances = fileOps.getDirectoriesSync(config.instanceDirectory);
} catch (e) {
// there are no instances created yet, probably missing the folder
instances = [];
}
return instances;
},
}
}
lib/link/index.js
+4
View File
@@ -0,0 +1,4 @@
module.exports = {
...require('lib/link/link'),
...require('lib/link/messages'),
};
lib/link/link.js
+399
View File
@@ -0,0 +1,399 @@
"use strict";
const events = require("events");
const io = require("socket.io-client");
const schema = require("lib/schema");
const errors = require("lib/errors");
const messages = require("lib/link/messages");
// Some definitions for the terminology used here:
// link: Either side of a master - client connection
// connection: the master server side of a link
// client: the side that is not the master server of a link
// message: the complete object sent using the 'message' event
// data: the data property of a message, essentially the payload.
// type: the type property of a message, identifies what to expect in data.
/**
* Common interface for server and client connections
*/
class Link {
constructor(source, target, socket) {
this.source = source;
this.target = target;
this.socket = socket;
this._seq = 1;
this._waiters = new Map();
this._handlers = new Map();
this._validators = new Map();
}
/**
* Process a received message on the link
*
* Validates and invokes the handler and/or waiters for a message that has
* been received. An unhandled message is considered to be an error.
*
* @throws {InvalidMessage}
* if validation failed, is missing, the message is invalid, or no
* handlers/waiters were invoked to handle the message.
*/
processMessage(message) {
if (!schema.message(message)) {
throw new errors.InvalidMessage("Malformed message");
}
let validator = this._validators.get(message.type);
if (!validator) {
throw new errors.InvalidMessage(`No validator for ${message.type}`);
}
if (!validator(message)) {
// TODO better logging of errors encountered validating
console.error(validator.errors);
throw new errors.InvalidMessage(`Validation failed for ${message.type}`);
}
let hadHandlers = this._processHandler(message);
let hadWaiters = this._processWaiters(message);
if (!hadWaiters && !hadHandlers) {
// XXX console.error(`Unhandled message: ${JSON.stringify(message, null, 4)}`);
throw new errors.InvalidMessage(`Unhandled message ${message.type}`);
}
}
/**
* Invoke the handler for this message if it exists
*
* @returns {Boolean} true if a handler was invoked.
*/
_processHandler(message) {
let handler = this._handlers.get(message.type);
if (handler) {
handler(message);
return true;
} else {
return false;
}
}
/**
* Invokes the waiters for this message if any
*
* @returns {Boolean} true if a waiter was triggered.
*/
_processWaiters(message) {
let { seq, type, data } = message;
let waiters = this._waiters.get(type);
if (!waiters || !waiters.length) {
return false;
}
let matched = [];
waitersLoop: for (let index = 0; index < waiters.length; index++) {
let waiter = waiters[index];
for (let [key, value] of Object.entries(waiter.data)) {
if (data[key] !== value) {
continue waitersLoop;
}
}
waiter.resolve(message);
matched.push(index);
}
matched.reverse();
for (let index of matched) {
waiters.splice(index, 1);
}
return matched.length > 0;
}
/**
* Set handler for a message type
*
* Set function called upon receiving a message of the given type.
*
* @param {string} type - Type of message to listen to
* @param {function} handler -
* Callack to invoke on messages of the given type.
* @param {function} validator -
* Function validating the message have the correct format.
*/
setHandler(type, handler, validator) {
if (this._handlers.has(type)) {
throw new Error(`${type} already has a handler`);
}
if (!validator) {
throw new Error("validator is required");
}
this.setValidator(type, validator);
this._handlers.set(type, handler);
}
/**
* Set validator for a message type
*
* Set function called upon receiving a message of the given type in
* order to validate the message. The function should return true if
* the message is valid and false otherwise.
*
* @param {string} type - Type of message to validate
* @param {function} validator - Validation function.
*/
setValidator(type, validator) {
if (this._validators.has(type)) {
throw new Error(`${type} already has a validator`);
}
this._validators.set(type, validator);
}
/**
* Send a message over the link
*
* @returns the sequence number of the message sent
*/
send(type, data) {
this.socket.send({ seq: this._seq, type, data });
return this._seq++;
}
/**
* Wait for a message matching given type and data
*
* Waits for a message over the link that matches the type given as well
* as any properties specified in data. For example specifying { foo:
* 21 } as the data will cause it to wait until a message with the right
* type and a foo property value of 21 in the data payload.
*/
async waitFor(type, data) {
if (!this._validators.has(type)) {
throw new Error(`no validator for ${type}`);
}
return new Promise((resolve, reject) => {
if (!this._waiters.has(type)) {
this._waiters.set(type, [{ data, resolve }]);
} else {
this._waiters.get(type).push({ data, resolve });
}
});
}
/**
* Close the connection with the given reason.
*
* Sends a close message and disconnects from the socket.
*/
close(reason) {
this.send('close', { reason });
this.disconnect();
}
/**
* Disconnect from the socket
*
* Abstract function that subclases override to implement disconnection
* from the socket.
*/
disconnect() {
throw new Error("Abstract function");
}
}
class Connection extends Link {
constructor(target, socket) {
super('master', target, socket);
this._seq = 3; // XXX this is private to Link
this.socket.on('message', payload => {
try {
this.processMessage(payload);
} catch (err) {
if (err instanceof errors.InvalidMessage) {
this.close(`Invalid message: ${err.message}`);
} else {
throw err;
}
}
});
this.socket.on('disconnect', this.disconnect.bind(this));
this.setHandler('close', payload => {
let address = this.socket.handshake.address
console.log(`SOCKET | received close from control ${address}: ${payload.data.reason}`);
this.socket.disconnect(true);
}, schema.close);
socket.send({ seq: 2, type: 'ready', data: {} });
}
disconnect() {
this.socket.disconnect(true);
}
}
class Client extends Link {
constructor(source, url, token) {
super(source, 'master', null);
this._url = url;
this._token = token;
this._state = "new";
this._events = new events.EventEmitter();
this.setHandler('close', message => {
console.log(`SOCKET | received close from server: ${message.data.reason}`);
this.socket.close();
return; // XXX what now?
}, schema.close);
}
_check(expectedStates) {
if (!expectedStates.includes(this._state)) {
throw new Error(
`Expected state ${expectedStates} but state is ${this._state}`
);
}
}
async connect() {
this._check(["new"]);
this._state = "handshake";
// Open socket.io connection to master
console.log(`SOCKET | connecting to ${this._url}`)
let url = new URL(this._url);
url.searchParams.append('token', this._token);
// I really have to question the choice of having the path connected
// to be passed as separate parameter instead of the namespace.
let urlPath = url.pathname + "socket.io";
url.pathname = "";
this.socket = io(url.href, {
path: urlPath,
reconnectionAttempts: 2,
// For now we do not verify TLS certificates since the default setup is
// to create a self-signed certificate.
rejectUnauthorized: false,
});
_attachSocketHandlers();
// Wait for the connection to be ready
await events.once(this._events, 'ready');
}
_attachSocketHandlers() {
this.socket.on("error", err => {
console.error("SOCKET | Error:", err);
this.socket.close();
});
/*this.socket.on("connect_error", err => {
console.log("SOCKET | Connect error:", err);
});*/
this.socket.on("connect_timeout", () => {
console.log("SOCKET | Connect timeout");
});
this.socket.on("reconnect", attempt => {
console.log(`SOCKET | Reconnect no. ${attempt} succeeded`);
});
this.socket.on("reconnect_attempt", () => {
console.log("SOCKET | Attempting reconnect");
});
this.socket.on("reconnecting", attempt => {
console.log(`SOCKET | Attempting reconnect no. ${attempt}`);
});
/*this.socket.on("reconnect_error", err => {
console.log("SOCKET | Reconnect error:", err);
});*/
this.socket.on("reconnect_failed", () => {
console.log("SOCKET | Reconnecting failed");
});
/*this.socket.on("ping", () => {
console.log("SOCKET | Ping");
});
this.socket.on("pong", (ms) => {
// XXX Latency might be useful
console.log(`SOCKET | Pong latency ${ms}ms`);
});*/
this.socket.on("connect", () => {
console.log("SOCKET | connected");
});
// Handle messages
this.socket.on("message", message => {
if (this._state === "handshake") {
this._processHandshake(message);
} else if (this._state === "ready") {
try {
this.processMessage(message);
} catch (err) {
if (err instanceof errors.InvalidMessage) {
this.close(`Invalid message: ${err.message}`);
} else {
throw err;
}
}
} else {
throw new Error(`Received message in unexpected state ${this._state}`);
}
})
}
disconnect(reason) {
this.socket.close();
// XXX what now?
}
_processHandshake(message) {
if (!schema.serverHandshake(message)) {
console.log("SOCKET | closing after received invalid handshake:", message);
this.close("Invalid handshake");
return;
}
let { seq, type, data } = message;
if (type == 'hello') {
console.log(`SOCKET | received hello from master version ${data.version}`);
this.register();
} else if (type === 'ready') {
console.log("SOCKET | received ready from master");
this._state = "ready";
this._events.emit('ready');
} else if (type === 'close') {
console.log(`SOCKET | received close from server: ${data.reason}`);
this._events.emit('error', new Error(`server closed during handshake: ${data.reason}`));
this.socket.close();
}
}
/**
* Register the link with the server
*
* This function is expected to be overriden by sub-classes and send the
* register message over the socket when invoked.
*/
register() {
throw Error("Abstract function");
}
}
module.exports = {
Link,
Connection,
Client,
}
lib/link/messages.js
+429
View File
@@ -0,0 +1,429 @@
"use strict";
const schema = require("lib/schema");
const errors = require("lib/errors");
/**
* Represents a request sent over the link
*
* A request-response message that can be innitated by either party of the
* link.
*/
class Request {
constructor({
type, sources, targets, requestProperties = {}, responseProperties = {}
}) {
this.type = type;
this.sources = sources;
this.targets = targets;
this.requestType = type + '_request';
this.responseType = type + '_response';
this._requestValidator = schema.compile({
$schema: "http://json-schema.org/draft-07/schema#",
properties: {
"type": { const: this.requestType },
"data": {
additionalProperties: false,
required: Object.keys(requestProperties),
properties: requestProperties,
},
},
});
this._responseValidator = schema.compile({
$schema: "http://json-schema.org/draft-07/schema#",
properties: {
"type": { const: this.responseType },
"data": {
anyOf: [
{
additionalProperties: false,
required: ["seq", ...Object.keys(responseProperties)],
properties: {
"seq": { type: "integer" },
...responseProperties,
},
},
{
additionalProperties: false,
required: ["seq", "error"],
properties: {
"seq": { type: "integer" },
"error": { type: "string" },
},
},
],
},
}
});
}
/**
* Attach to a link
*
* Set a the validator for the response on the given link if it's a
* source and sets the handler for the request if it is a target.
*
* Does nothing if the link is neither a source nor a target for this
* request.
*
* @param {Link} link - The link to attach to.
* @param {Function} handler -
* Async function to invoke with link set as this to handle the
* request request. Only used on the target side.
* @throws {Error} if the handler is needed and not defined.
*/
attach(link, handler) {
// Source side.
if (this.sources.includes(link.source) && this.targets.includes(link.target)) {
link.setValidator(this.responseType, this._responseValidator);
}
// Target side. Note: Source and target is reversed here.
if (this.targets.includes(link.source) && this.sources.includes(link.target)) {
if (!handler) {
throw new Error(`Missing handler for ${this.requestType} on ${link.target}-${link.source} link`);
}
link.setHandler(this.requestType, message => {
handler.call(link, message).then(response => {
if (response === undefined) {
// XXX Should we allow implicit responses like this?
response = {};
}
if (Object.prototype.hasOwnProperty.call(response, 'seq')) {
throw new errors.RequestError("response contains reserved property 'seq'");
}
link.send(this.responseType, { seq: message.seq, ...response });
}).catch(err => {
if (!(err instanceof errors.RequestError)) {
console.error(`Unexpected error while responding to ${this.requestType}`, err);
}
link.send(this.responseType, { seq: message.seq, error: err.message });
});
}, this._requestValidator);
}
}
/**
* Send request over the given link
*
* Sends the given data over the link as the request payload and waits
* for the response.
*
* @returns {object} response data
*/
async send(link, data = {}) {
// XXX validate link target/source?
let seq = link.send(this.requestType, data);
let responseMessage = await link.waitFor(this.responseType, { seq });
if (responseMessage.data.error) {
throw new errors.RequestError(responseMessage.data.error);
}
return responseMessage.data;
}
}
/**
* Represents a request sent to the master that is forwarded to an instance
*
* This is a two part request with a master part and a slave part. The
* master part is done by the forwardInstanceRequest method on the master
* side of link which finds the slave the requuested instance is on and
* forwards it there using the foward method.
*
* The slave part is done by the forwardInstanceRequest on the slave side of
* the link and finds the instance and then calls the handler with it and
* the message.
*/
class InstanceRequest {
constructor({
type, sources, requestProperties = {}, responseProperties = {}
}) {
this._masterRequest = new Request({
type: type + '_master',
sources,
targets: ['master'],
requestProperties: {
"instance_id": { type: "integer" },
...requestProperties,
},
responseProperties,
});
this._slaveRequest = new Request({
type: type + '_slave',
sources: ['master'],
targets: ['slave'],
requestProperties: {
"instance_id": { type: "integer" },
...requestProperties,
},
responseProperties,
});
}
/**
* Attach to a link
*
* Attach master/slave part of the instance request. If this is the
* slave side the handler argument is required.
*
* Does nothing if the link is neither a source nor a target for the
* slave or master request.
*
* @param {Link} link - The link to attach to.
* @param {Function} handler -
* Async function to to handle the instance request. It is passed on
* to forwardInstanceRequest on the slave side.
* @throws {Error} if the handler is needed and not defined.
*/
attach(link, handler) {
this._masterRequest.attach(link, async (message) => {
return await link.forwardInstanceRequest(message, this);
});
this._slaveRequest.attach(link, async (message) => {
return await link.forwardInstanceRequest(message, handler);
});
}
/**
* Forward the message over the link
*
* This method is expected to be called by forwardInstanceRequest on the
* master with the slave link that the instance is on.
*/
async forward(link, message) {
let response = await this._slaveRequest.send(link, message.data);
delete response.seq;
return response;
}
/**
* Send request over the given link
*
* Sends the given data over the link to the master and waits for it to
* forward the request to the slave with the instance on and return the
* result of it.
*
* @returns {object} response data
*/
async send(link, data = {}) {
return await this._masterRequest.send(link, data);
}
}
let requests = {}
// Management requests
requests.listSlaves = new Request({
type: 'list_slaves',
sources: ['control'],
targets: ['master'],
responseProperties: {
"list": {
type: "array",
items: {
additionalProperties: false,
required: ["agent", "version", "name", "id", "connected"],
properties: {
"agent": { type: "string" },
"version": { type: "string" },
"name": { type: "string" },
"id": { type: "integer" },
"connected": { type: "boolean" },
},
},
},
},
});
requests.listInstances = new Request({
type: 'list_instances',
sources: ['control'],
targets: ['master'],
responseProperties: {
"list": {
type: "array",
items: {
additionalProperties: false,
required: ["name", "id", "slave_id"],
properties: {
"name": { type: "string" },
"id": { type: "integer" },
"slave_id": { type: "integer" },
},
},
},
},
});
requests.createInstanceCommand = new Request({
type: 'create_instance',
sources: ['control'],
targets: ['master'],
requestProperties: {
"name": { type: "string" },
"slave_id": { type: "integer" },
},
});
requests.createInstance = new Request({
type: 'create_instance',
sources: ['master'],
targets: ['slave'],
requestProperties: {
"id": { type: "integer" },
"options": {
additionalProperties: false,
required: [
"name", "description", "visibility", "username", "token", "game_password",
"verify_user_identity", "allow_commands", "auto_pause",
],
properties: {
"name": { type: "string" },
"description": { type: "string" },
"visibility": { type: "object" },
"username": { type: "string" },
"token": { type: "string" },
"game_password": { type: "string" },
"verify_user_identity": { type: "boolean" },
"allow_commands": { type: "string" },
"auto_pause": { type: "boolean" },
},
},
},
});
requests.createSave = new InstanceRequest({
type: 'create_save',
sources: ['control'],
});
requests.startInstance = new InstanceRequest({
type: 'start_instance',
sources: ['control'],
});
requests.stopInstance = new InstanceRequest({
type: 'stop_instance',
sources: ['control'],
});
requests.deleteInstance = new InstanceRequest({
type: 'delete_instance',
sources: ['control'],
});
requests.sendRcon = new InstanceRequest({
type: 'send_rcon',
sources: ['control'],
requestProperties: {
"command": { type: "string" },
},
responseProperties: {
"result": { type: "string" },
}
});
class Event {
constructor({ type, sources, targets, eventProperties = {} }) {
this.type = type;
this.sources = sources;
this.targets = targets;
this.eventType = type + '_event';
this._eventValidator = schema.compile({
$schema: "http://json-schema.org/draft-07/schema#",
properties: {
"type": { const: this.eventType },
"data": {
additionalProperties: false,
required: Object.keys(eventProperties),
properties: eventProperties,
},
},
});
}
attach(link, handler) {
// Target side. Note: Source and target is reversed here.
if (this.targets.includes(link.source) && this.sources.includes(link.target)) {
if (!handler) {
throw new Error(`Missing handler for ${this.eventType} on ${link.target}-${link.source} link`);
}
link.setHandler(this.eventType, message => {
// XXX Should event handlers even be allowed to be async?
handler.call(link, message).catch(err => {
console.error(`Unexpected error while handling ${this.eventType}`, err);
});
}, this._eventValidator);
}
}
/**
* Send event over the given link
*
* Sends the given event data over the link.
*/
send(link, data = {}) {
link.send(this.eventType, data);
}
}
let events = {};
events.updateInstances = new Event({
type: 'update_instances',
sources: ['slave'],
targets: ['master'],
eventProperties: {
"instances": {
type: "array",
items: {
type: "object",
additionalProperties: false,
required: ["id", "name"],
properties: {
"id": { type: "integer" },
"name": { type: "string" },
},
},
},
},
});
/**
* Attaches all requests and events to the given link
*
* Loops through all requests and events defined and attaches all of
* them to the link. For requests the handler used is looked up on the
* link instance as the name of the request + 'RequestHandler'. For
* events the the handler used is looked up on the link instance as the
* name of the event + 'EventHandler'.
*/
function attachAllMessages(link) {
for (let [name, request] of Object.entries(requests)) {
request.attach(link, link[name + 'RequestHandler']);
}
for (let [name, event] of Object.entries(events)) {
event.attach(link, link[name + 'EventHandler']);
}
}
module.exports = {
Request,
InstanceRequest,
Event,
attachAllMessages,
requests,
events,
}
lib/manager/modManager.js
+1 -2
View File
@@ -36,8 +36,7 @@ modManager.listMods(config, instance);
*/
async function listMods(config, instance) {
return new Promise((resolve, reject) => {
const instances = require("./../getInstances")(config).sync();
if (instances.includes(instance.name)) {
if (true) {
factorioAPI.init(false, instance.path("instanceMods"), instance.path("saves"), instance.server.version);
factorioAPI.authenticate({
username: config.username,
lib/schema.js
+144
View File
@@ -0,0 +1,144 @@
const ajv = new (require('ajv'))({
verbose: true,
format: "full",
extendRefs: "fail",
strictDefaluts: true,
strictKeywords: true,
})
const closeSchema = {
$schema: "http://json-schema.org/draft-07/schema#",
title: "Close Message",
description: "Format for a close messages on a link",
properties: {
"type": { const: "close" },
"data": {
additionalProperties: false,
required: ["reason"],
properties: {
"reason": { type: "string" },
}
},
},
}
const close = ajv.compile(closeSchema);
/**
* The base format for any message sent over the link
*/
const messageSchema = {
$schema: "http://json-schema.org/draft-07/schema#",
title: "Message",
description: "Format for messages on a link",
type: "object",
additionalProperties: false,
required: ["seq", "type", "data"],
properties: {
"seq": { type: "integer" },
"type": { type: "string" },
"data": { type: "object" },
},
}
const message = ajv.compile(messageSchema);
const serverActions = [
{},
];
const actionResponse = {
additionalProperties: false,
required: ["action", "id", "status", "message"],
properties: {
"action": { enum: ["TODO"] },
"id": { type: "integer" },
"status": { enum: ["started", "done", "error"] },
"message": { type: "string" },
},
};
const serverHandshake = ajv.compile({
$schema: "http://json-schema.org/draft-07/schema#",
title: "Server Handshake",
description: "Message sent from the server to a client during the handshake",
allOf: [ messageSchema ],
anyOf: [
{
properties: {
"type": { const: "hello" },
"data": {
additionalProperties: false,
required: ["version"],
properties: {
"version": { type: "string" },
}
},
},
},
{
properties: {
"type": { const: "ready" },
"data": {
additionalProperties: false,
/* XXX anything here?
required: ["version"],
properties: {
"version": { type: "string" },
}*/
},
},
},
closeSchema,
],
});
const clientHandshake = ajv.compile({
$schema: "http://json-schema.org/draft-07/schema#",
title: "Client Message",
description: "Message sent from a client to the server",
allOf: [ messageSchema ],
anyOf: [
{
properties: {
"type": { const: "register_slave" },
"data": {
additionalProperties: false,
required: ["agent", "version", "name", "id"],
properties: {
"agent": { type: "string" },
"version": { type: "string" },
"name": { type: "string" },
"id": { type: "integer" },
},
},
},
},
{
properties: {
"type": { const: "register_control" },
"data": {
additionalProperties: false,
required: ["agent", "version"],
properties: {
"agent": { type: "string" },
"version": { type: "string" },
}
},
},
},
closeSchema,
],
});
module.exports = {
compile: ajv.compile.bind(ajv),
message,
close,
serverHandshake,
clientHandshake,
}
master.js
+223 -196
View File
@@ -24,6 +24,7 @@ const crypto = require('crypto');
const base64url = require('base64url');
const moment = require("moment");
const request = require("request");
const version = require("./package").version;
// constants
let config = {};
@@ -33,6 +34,9 @@ const generateSSLcert = require("lib/generateSSLcert");
const pluginManager = require("lib/manager/pluginManager");
const database = require("lib/database");
const factorio = require("lib/factorio");
const schema = require("lib/schema");
const link = require("lib/link");
const errors = require("lib/errors");
// homemade express middleware for token auth
const authenticate = require("lib/authenticate");
@@ -155,10 +159,22 @@ function registerMoreMetrics(){
// set up database
const db = {};
async function loadDatabase(databaseDirectory) {
let slavesPath = path.resolve(databaseDirectory, "slaves.json");
console.log(`Loading ${slavesPath}`);
db.slaves = await database.loadJsonAsMap(slavesPath);
/**
* Load Map from JSON file in the database directory.
*/
async function loadMap(databaseDirectory, file) {
let databasePath = path.resolve(databaseDirectory, file);
console.log(`Loading ${databasePath}`);
return await database.loadJsonArrayAsMap(databasePath);
}
/**
* Save Map to JSON file in the database directory.
*/
async function saveMap(databaseDirectory, file, map) {
let databasePath = path.resolve(config.databaseDirectory, file);
console.log(`Saving ${databasePath}`);
await database.saveMapAsJsonArray(databasePath, map);
}
@@ -167,14 +183,8 @@ async function shutdown() {
console.log('Ctrl-C...');
let exitStartTime = Date.now();
try {
// set insane limit to slave length, if its longer than this we are probably being ddosed or something
if (db.slaves && db.slaves.size < 2000000) {
let file = path.resolve(config.databaseDirectory, "slaves.json");
console.log(`writing ${file}`);
await database.saveMapAsJson(file, db.slaves);
} else if (db.slaves) {
console.error(`Slave database too large, not saving (${db.slaves.size}`);
}
await saveMap(config.databaseDirectory, "slaves.json", db.slaves);
await saveMap(config.databaseDirectory, "instances.json", db.instances);
for(let i in masterPlugins){
let plugin = masterPlugins[i];
@@ -193,135 +203,6 @@ async function shutdown() {
}
}
/**
POST World ID management. Slaves post here to tell the server they exist
@memberof clusterioMaster
@instance
@alias /api/getID
*/
app.post("/api/getID", authenticate.middleware, function(req,res) {
endpointHitCounter.labels(req.route.path).inc();
// request.body should be an object
// {rconPort, rconPassword, serverPort, mac, time}
// time us a unix timestamp we can use to check for how long the server has been unresponsive
// we should save that somewhere and give appropriate response
if(req.body){
if (!db.slaves.has(String(req.body.unique))) {
db.slaves.set(String(req.body.unique), {});
}
let slave = db.slaves.get(String(req.body.unique));
for(let k in req.body){
if(k != "meta" && req.body.hasOwnProperty(k)){
slave[k] = req.body[k];
}
}
// console.log("Slave registered: " + slave.mac + " : " + slave.serverPort+" at " + slave.publicIP + " with name " + slave.instanceName);
}
});
/**
POST Allows you to add metadata related to slaves for other tools, like owner data, descriptions or statistics.
@memberof clusterioMaster
@instance
@alias /api/editSlaveMeta
*/
app.post("/api/editSlaveMeta", authenticate.middleware, function(req,res) {
endpointHitCounter.labels(req.route.path).inc();
// request.body should be an object
// {instanceID, pass, meta:{x,y,z}}
if(req.body && req.body.instanceID && req.body.meta){
if (db.slaves.has(String(req.body.instanceID))) {
let slave = db.slaves.get(String(req.body.instanceID));
if(!slave.meta) {
slave.meta = {};
}
slave.meta = deepmerge(slave.meta, req.body.meta, {clone:true});
let metaPortion = JSON.stringify(req.body.meta);
if(metaPortion.length > 50) {
metaPortion = metaPortion.substring(0,20) + "...";
}
// console.log("Updating slave "+slave.instanceName+": " + slave.mac + " : " + slave.serverPort+" at " + slave.publicIP, metaPortion);
res.sendStatus(200);
} else {
res.send("ERROR: Invalid instanceID or password")
}
}
});
/**
POST Get metadata from a single slave. For whenever you find the data returned by /api/slaves overkill.
@memberof clusterioMaster
@instance
@alias /api/getSlaveMeta
*/
app.post("/api/getSlaveMeta", function (req, res) {
endpointHitCounter.labels(req.route.path).inc();
console.log("body", req.body);
if(req.body && req.body.instanceID){
if (db.slaves.has(String(req.body.instanceID))) {
let slave = db.slaves.get(String(req.body.instanceID));
console.log("returning meta for ", req.body.instanceID);
if(!slave.meta) {
slave.meta = {};
}
res.send(JSON.stringify(slave.meta))
} else {
res.status(404);
res.send('{"status": 404, "info": "Slave not registered"}')
}
} else {
res.status(400);
res.send('{"INVALID_REQUEST":1}');
}
});
/**
Prepare the slaveCache for both /api/slaves calls
*/
let slaveCache = {
timestamp: Date.now(),
};
function getSlaves() {
if(!slaveCache.cache || Date.now() - slaveCache.timestamp > 5000) {
let copyOfSlaves = JSON.parse(JSON.stringify(database.mapToObject(db.slaves)));
slaveCache.cache = copyOfSlaves;
slaveCache.timestamp = Date.now();
}
return slaveCache;
}
/**
GET endpoint for getting information about all our slaves
@memberof clusterioMaster
@instance
@alias /api/slaves
*/
app.get("/api/slaves", function(req, res) {
endpointHitCounter.labels(req.route.path).inc();
res.header("Access-Control-Allow-Origin", "*");
res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
let slaveCache = getSlaves();
for(key in slaveCache.cache) {
slaveCache.cache[key].rconPassword = "hidden";
}
res.send(slaveCache.cache);
});
/**
POST endpoint for getting the rconPasswords of all our slaves
@memberof clusterioMaster
@instance
@alias /api/rconPasswords
*/
app.post("/api/rconPasswords", authenticate.middleware, function(req, res) {
endpointHitCounter.labels(req.route.path).inc();
res.header("Access-Control-Allow-Origin", "*");
res.header("Access-Control-Allow-Headers", "Origin, X-Requested-With, Content-Type, Accept");
let slaveCache = getSlaves();
let validKeys = ['rconPort', 'rconPassword', 'publicIP'];
for(key in slaveCache.cache) {
slaveCache.cache[key] = {'publicIP': slaveCache.cache[key].publicIP, 'rconPort': slaveCache.cache[key].rconPort, 'rconPassword': slaveCache.cache[key].rconPassword};
}
res.send(slaveCache.cache);
});
// wrap a request in an promise
async function downloadPage(url) {
return new Promise((resolve, reject) => {
@@ -361,49 +242,170 @@ app.get("/api/getFactorioLocale", function(req,res){
}
});
/* Websockets for plugins */
// socket.io connection for slaves
var io = require("socket.io")({});
const ioMetrics = require("socket.io-prometheus");
ioMetrics(io);
/* Websockets for send and recieve combinators */
var wsSlaves = {};
class wsSlave {
constructor(instanceID, socket){
this.instanceID = instanceID;
this.socket = socket;
this.lastBeat = Date.now();
class BaseConnection extends link.Connection {
constructor(target, socket) {
super(target, socket);
link.attachAllMessages(this);
}
this.socket.on("heartbeat", () => {
prometheusWsUsageCounter.labels('heartbeat', this.instanceID).inc();
this.lastBeat = Date.now();
});
async forwardInstanceRequest(message, request) {
let instance = db.instances.get(message.data.instance_id);
if (!instance) {
throw new errors.RequestError(`Instance with ID ${instance_id} does not exist`);
}
this.socket.on("gameChat", data => {
prometheusWsUsageCounter.labels('gameChat', this.instanceID).inc();
if(typeof data === "object"){
data.timestamp = Date.now();
if(!global.wsChatRecievers) global.wsChatRecievers = [];
global.wsChatRecievers.forEach(socket => {
socket.emit("gameChat", data);
});
let connection = slaveConnections.get(instance.slaveId);
if (!connection) {
throw new errors.RequestError("Slave containing instance is not connected");
}
return await request.forward(connection, message);
}
}
let controlConnections = new Array();
class ControlConnection extends BaseConnection {
constructor(registerData, socket) {
super('control', socket)
this._agent = registerData.agent;
this._version = registerData.version;
socket.on('disconnect', () => {
let index = controlConnections.indexOf(this);
if (index !== -1) {
controlConnections.splice(index, 1);
}
});
}
this.socket.on("alert", alert => {
prometheusWsUsageCounter.labels('alert', this.instanceID).inc();
if(typeof alert === "object"){
alert.timestamp = Date.now();
if(!global.wsAlertRecievers) global.wsAlertRecievers = [];
global.wsAlertRecievers.forEach(socket => {
socket.emit("alert", alert);
});
async listSlavesRequestHandler(message) {
let list = [];
for (let slave of db.slaves.values()) {
list.push({
agent: slave.agent,
version: slave.version,
id: slave.id,
name: slave.name,
connected: slaveConnections.has(slave.id),
});
}
return { list };
}
async listInstancesRequestHandler(message) {
let list = [];
for (let instance of db.instances.values()) {
list.push({
id: instance.id,
name: instance.name,
slave_id: instance.slaveId,
});
}
return { list };
}
// XXX should probably add a hook for slave reuqests?
async createInstanceCommandRequestHandler(message) {
let { slave_id, name } = message.data;
let connection = slaveConnections.get(slave_id);
if (!connection) {
throw new errors.RequestError("Slave is not connected");
}
await link.requests.createInstance.send(connection, {
id: Math.random() * 2**31 | 0, // TODO: add id option
options: {
'name': name,
'description': config.description,
'visibility': config.visibility,
'username': config.username,
'token': config.token,
'game_password': config.game_password,
'verify_user_identity': config.verify_user_identity,
'allow_commands': config.allow_commands,
'auto_pause': config.auto_pause,
}
});
}
}
var slaveConnections = new Map();
class SlaveConnection extends BaseConnection {
constructor(registerData, socket) {
super('slave', socket);
this._agent = registerData.agent;
this._id = registerData.id;
this._name = registerData.name;
this._version = registerData.version;
db.slaves.set(this._id, {
agent: this._agent,
id: this._id,
name: this._name,
version: this._version,
});
socket.on('message', () => {
prometheusWsUsageCounter.labels('message', "other").inc();
});
socket.on('disconnect', () => {
if (slaveConnections.get(this._id) === this) {
slaveConnections.delete(this._id);
}
// XXX for backwards compatibility only
if (global.wsChatRecievers) {
global.wsChatRecievers.delete(this._id);
}
});
// XXX for backwards compatibility only
socket.on("gameChat", data => {
prometheusWsUsageCounter.labels('gameChat', this.instanceID).inc();
if(typeof data === "object"){
data.timestamp = Date.now();
if(!global.wsChatRecievers) global.wsChatRecievers = new Map();
for (let slave of global.wsChatRecievers.values()) {
slave.socket.emit("gameChat", data);
}
}
});
socket.on("registerChatReciever", data => {
prometheusWsUsageCounter.labels("registerChatReciever", "other").inc();
if(!global.wsChatRecievers) global.wsChatRecievers = new Map();
global.wsChatRecievers.set(this._id, this);
});
}
async updateInstancesEventHandler(message) {
// Prune any instances previously had by the slave
for (let id of [...db.instances.keys()]) {
if (db.instances.get(id).slaveId === this._id) {
db.instances.delete(id);
}
}
for (let instance of message.data.instances) {
db.instances.set(instance.id, {
id: instance.id,
name: instance.name,
slaveId: this._id,
});
}
}
}
// Authentication middleware for socket.io connections
io.use((socket, next) => {
let token = socket.handshake.query.token;
authenticate.check(token).then((result) => {
@@ -416,42 +418,65 @@ io.use((socket, next) => {
});
});
/**
* Returns true if value is a signed 32-bit integer
*
* @param value - value to test.
* @returns {Boolean}
* true if value is an integer between -2**31 and 2**31-1.
*/
function isInteger(value) {
return value | 0 === value;
}
io.on('connection', function (socket) {
// cleanup dead sockets from disconnected people
let terminatedConnections = 0;
let currentConnections = Object.keys(wsSlaves).length;
[wsSlaves].forEach(list => {
Object.keys(list).forEach(connectionID => {
let connection = list[connectionID];
if(connection.lastBeat < (Date.now() - 30000)){
terminatedConnections++;
delete list[connectionID];
console.log(`SOCKET | new connection from ${socket.handshake.address}`);
// Start connection handshake
socket.send({ seq: 1, type: 'hello', data: { version }});
// Handle socket handshake
socket.once('message', (payload) => {
prometheusWsUsageCounter.labels('message', "other").inc();
if (!schema.clientHandshake(payload)) {
console.log(`SOCKET | closing ${socket.handshake.address} after receiving invalid handshake`, payload);
socket.send({ type: 'close', data: {
reason: "Invalid handshake",
}});
socket.disconnect(true);
return;
}
let { seq, type, data } = payload;
if (type === "register_slave") {
let connection = slaveConnections.get(data.id);
if (connection) {
console.log(`SOCKET | disconecting existing connection for slave ${data.id}`);
connection.close("Registered from another connection");
}
});
});
if(terminatedConnections > 0) console.log("SOCKET | There are currently "+currentConnections+" websocket connections, deleting "+terminatedConnections+" on timeout");
// tell our friend that we are listening
setTimeout(()=>socket.emit('hello', { hello: 'world' }), 5000);
console.log(`SOCKET | registered slave ${data.id} version ${data.version}`);
slaveConnections.set(data.id, new SlaveConnection(data, socket));
/* Websockets for send and recieve combinators */
socket.on("registerSlave", function(data) {
prometheusWsUsageCounter.labels('registerSlave', "other").inc();
if(data && data.instanceID){
wsSlaves[data.instanceID] = new wsSlave(data.instanceID, socket);
console.log("SOCKET | Created new wsSlave: "+ data.instanceID);
} else if (type === "register_control") {
console.log(`SOCKET | registered control from ${socket.handshake.address}`);
controlConnections.push(new ControlConnection(data, socket));
} else if (type === "close") {
console.log(`SOCKET | received close from ${socket.handshake.address}: ${data.reason}`);
socket.disconnect(true);
return;
}
});
socket.on("registerChatReciever", function(data){
prometheusWsUsageCounter.labels("registerChatReciever", "other").inc();
if(!global.wsChatRecievers) global.wsChatRecievers = [];
global.wsChatRecievers.push(socket);
});
// XXX TODO
/*
socket.on("registerAlertReciever", function(data){
prometheusWsUsageCounter.labels("registerAlertReciever", "other").inc();
if(!global.wsAlertRecievers) global.wsAlertRecievers = [];
global.wsAlertRecievers.push(socket);
});
});*/
});
// handle plugins on the master
@@ -556,7 +581,9 @@ async function startServer() {
config.databaseDirectory = args.databaseDirectory || config.databaseDirectory || "./database";
await fs.ensureDir(config.databaseDirectory);
await loadDatabase(config.databaseDirectory);
db.slaves = await loadMap(config.databaseDirectory, "slaves.json");
db.instances = await loadMap(config.databaseDirectory, "instances.json");
authenticate.setAuthSecret(config.masterAuthSecret);
// Make sure we're actually going to listen on a port
package.json
+2 -2
View File
@@ -45,7 +45,6 @@
"express-prometheus-request-metrics": "github:burnash/express-prometheus-request-metrics",
"follow-redirects": "^1.5.0",
"fs-extra": "^8.1.0",
"getmac": "^1.4.0",
"ini": "^1.3.5",
"jsonwebtoken": "^8.2.1",
"jszip": "^3.2.2",
@@ -67,7 +66,8 @@
"socket.io-prometheus": "^0.2.1",
"sync-request": "^4.0.1",
"tail": "^2.0.2",
"typescript-collections": "^1.3.2"
"typescript-collections": "^1.3.2",
"yargs": "^14.2.0"
},
"devDependencies": {
"html5-validator": "^1.2.1",
sharedPlugins/UPSdisplay/index.js
+1
View File
@@ -7,6 +7,7 @@ class UPSdisplay {
this.config = slaveConfig;
this.messageInterface = msgInterface;
this.socket = extras.socket;
this.slave = extras.slave;
this.historicalTicks = [];
setInterval(() => {
sharedPlugins/clusterioMod/index.js
-3
View File
@@ -21,9 +21,6 @@ module.exports = class remoteCommands {
this.messageInterface = messageInterface;
this.config = mergedConfig;
this.socket = extras.socket;
this.socket.on("hello", () => {
});
const that = this;
const instance = process.argv[3];
const instancedirectory = this.config.instanceDirectory + '/' + instance;
sharedPlugins/globalChat/index.js
+3 -3
View File
@@ -4,9 +4,9 @@ module.exports = class remoteCommands {
this.config = mergedConfig;
this.socket = extras.socket;
this.instances = {};
this.socket.on("hello", () => this.socket.emit("registerChatReciever"));
this.socket.emit("registerChatReciever");
this.socket.on("gameChat", async data => {
if(data.instanceID && data.instanceID != this.config.unique.toString()){
if(data.data.includes("[CHAT]")
sharedPlugins/inventoryImports/interfacingFunctions.js
-5
View File
@@ -140,11 +140,6 @@ function pollInventories(outputFile){
// Warning: This command is not in sync with lua commands.lua
return `/silent-command local a='{"players":{'local b=false;for c,d in pairs(game.players)do if d.connected and d.character then if b then a=a..','else b=true end;local e=game.players[c].get_inventory(defines.inventory.character_main).get_contents()local g={}for h=1,game.players[c].force.character_logistic_slot_count do g[h]=game.players[c].character.get_request_slot(h)end;a=a..'"'..c..'":{"inventory":{'local i=false;for j,k in pairs(e)do if i then a=a..','else i=true end;a=a..'"'..j..'":'..k end;a=a..'},"requestSlots":{'i=false;for j,k in pairs(g)do if i then a=a..','else i=true end;a=a..'"'..k['name']..'":'..k['count']end;a=a..'}}'end end;a=a.."}}"game.write_file("${outputFile}",a,true,0)`
/*
needle.post(config.masterURL+'/api/editSlaveMeta', {instanceID: config.unique, password: config.clientPassword, meta: {UPS:UPS}}, function(err, resp) {
// success?
});
*/
}
module.exports = {
test/client.js
+2 -2
View File
@@ -6,7 +6,7 @@ const client = require("../client");
describe("Client testing", function() {
describe("class Instance", function() {
let instance = new client._Instance("dir", "foo")
let instance = new client._Instance("dir", "factorioDir", { name:"foo" })
it("should give the name on .name", function() {
assert.equal(instance.name, "foo");
})
@@ -81,7 +81,7 @@ describe("Client testing", function() {
fs.outputFileSync(path.join(testDir, "shared", "mod_a.zip"), "a");
fs.outputFileSync(path.join(testDir, "shared", "mod_b.zip"), "b");
fs.outputFileSync(path.join(testDir, "shared", "mod.dat"), "c");
let instance = new client._Instance(path.join(testDir, "instance"), "test");
let instance = new client._Instance(path.join(testDir, "instance"), "factorioDir", { name: "test" });
fs.outputFileSync(instance.path("mods", "mod_i.zip"), "i");
let discardingLogger = {
test/integration/server.js
+3 -3
View File
@@ -50,9 +50,9 @@ describe("Integration of lib/factorio/server", function() {
it("contains the settings used by Clusterio", async function() {
let keysUsed = new Set([
"name", "description", "tags", "max_players", "visibility", "username", "token",
"game_password", "require_user_verification", "allow_commands", "autosave_interval",
"autosave_slots", "afk_autokick_interval", "auto_pause",
"name", "description", "tags", "visibility", "username", "token",
"game_password", "require_user_verification", "allow_commands",
"auto_pause",
]);
for (let key of Object.keys(settings)) {
test/lib/link/link.js
+337
View File
@@ -0,0 +1,337 @@
const assert = require('assert').strict;
const events = require('events');
const link = require("lib/link");
const errors = require("lib/errors");
const schema = require("lib/schema");
class MockSocket {
constructor() {
this.sentMessages = [];
this.events = new Map();
this.handshake = { address: "socket.test" };
}
send(message) {
this.sentMessages.push(message);
}
on(event, fn) {
this.events.set(event, fn);
}
disconnect() {
this.disconnectCalled = true;
}
close() {
this.closeCalled = true;
}
};
describe("lib/link/link", function() {
describe("class Link", function() {
let testLink = new link.Link('source', 'target', new MockSocket);
testLink.setValidator('simple', schema.compile({
properties: {
"data": {
type: 'object',
properties: {
"string": { type: 'string' },
},
},
},
}));
describe(".setValidator()", function() {
it("should set the validator", function() {
testLink.setValidator('test', () => true);
assert(testLink._validators.has('test'), "validator was not added");
});
it("should throw on duplicated validator", function() {
assert.throws(
() => testLink.setValidator('test', () => true),
new Error("test already has a validator")
);
});
});
describe(".setHandler()", function() {
it("should throw if validator is not passed", function() {
assert.throws(
() => testLink.setHandler('no_validator_test', (message) => {}),
new Error("validator is required")
);
});
it("should set the handler", function() {
let handler = (message) => {};
testLink.setHandler('handler_test', handler, (message) => true);
assert.equal(testLink._handlers.get('handler_test'), handler);
});
it("should throw on duplicated handler", function() {
assert.throws(
() => testLink.setHandler('handler_test', (message) => {}),
new Error("handler_test already has a handler")
);
});
});
describe(".processMessage()", function() {
it("should throw on invalid message", function() {
assert.throws(
() => testLink.processMessage({ data: "invalid message" }),
new errors.InvalidMessage("Malformed message")
);
});
it("should throw on message without validator", function() {
assert.throws(
() => testLink.processMessage({ seq: 1, type: 'no_validator', data: {} }),
new errors.InvalidMessage("No validator for no_validator")
);
});
it("should throw on message failing validation", function() {
assert.throws(
() => testLink.processMessage({ seq: 1, type: 'simple', data: { string: 1 }}),
new errors.InvalidMessage("Validation failed for simple")
);
});
it("should throw on unhandled message", function() {
assert.throws(
() => testLink.processMessage({ seq: 1, type: 'simple', data: { string: "a" }}),
new errors.InvalidMessage("Unhandled message simple")
);
});
});
describe("._processHandler()", function() {
it("should call the handler", function() {
let handled = [];
testLink.setHandler('handled', (message) => handled.push(message), (message) => true)
testLink._processHandler({ seq: 1, type: 'handled', data: { test: "foo" } });
assert.deepEqual(handled, [{ seq: 1, type: 'handled', data: { test: "foo" } }]);
});
it("should return true if handled", function() {
let result = testLink._processHandler({ seq: 1, type: 'handled', data: {} });
assert.equal(result, true);
});
it("should return false if not handled", function() {
let result = testLink._processHandler({ seq: 1, type: 'not_handled', data: {} });
assert.equal(result, false);
});
});
describe(".disconnect()", function() {
it("is abstract", function() {
assert.throws(
() => testLink.disconnect(),
new Error("Abstract function")
);
});
});
describe(".send()", function() {
it("calls send on the socket", function() {
let seq = testLink.send('test', { test: true });
assert.deepEqual(testLink.socket.sentMessages, [{ seq, type: 'test', data: { test: true }}]);
testLink.socket.sentMessages = [];
});
});
describe("._processWaiters()", function() {
it("should call waiters", async function() {
let waiter = new Promise(resolve => {
testLink._waiters.set('test', [{ resolve, data: {} }]);
});
assert(
testLink._processWaiters({ type: 'test', seq: 4, data: {} }),
"Waiter was not processed"
);
assert.deepEqual(await waiter, { type: 'test', seq: 4, data: {} });
});
it("should filter on data", async function() {
let waiter = new Promise(resolve => {
testLink._waiters.set('test', [{ resolve, data: { a: 1, b: 2 } }]);
});
assert(
!testLink._processWaiters({ type: 'test', seq: 4, data: {} }),
"Incorrectly matched empty message"
);
assert(
!testLink._processWaiters({ type: 'test', seq: 4, data: { a: 1, b: 0 } }),
"Incorrectly matched message with wrong data"
);
assert(
testLink._processWaiters({ type: 'test', seq: 4, data: { a: 1, b: 2, c: 3 } }),
"Did not match correct message"
);
assert.deepEqual(await waiter, { type: 'test', seq: 4, data: { a: 1, b: 2, c: 3 } });
});
});
describe(".waitFor()", function() {
testLink.setValidator('waiter', (message) => true);
it("should throw on missing validator", async function() {
await assert.rejects(
testLink.waitFor('no_validator', {}),
new Error("no validator for no_validator")
);
});
it("should wait for a given message", async function() {
let result = testLink.waitFor('waiter', {});
assert(
testLink._processWaiters({ type: 'waiter', seq: 5, data: {} }),
"Did not proccess waiter"
);
assert.deepEqual(await result, { type: 'waiter', seq: 5, data: {} });
});
it("should handle multiple waiters", async function() {
let result1 = testLink.waitFor('waiter', {});
let result2 = testLink.waitFor('waiter', {});
assert(
testLink._processWaiters({ type: 'waiter', seq: 5, data: {} }),
"Did not proccess waiter"
);
assert.deepEqual(await result1, { type: 'waiter', seq: 5, data: {} });
assert.deepEqual(await result2, { type: 'waiter', seq: 5, data: {} });
});
});
describe(".close()", function() {
it("should send close and call disconnect", function() {
let called = false;
testLink.disconnect = () => { called = true; }
testLink.close("test reason");
assert.deepEqual(
testLink.socket.sentMessages,
[{ seq: testLink._seq - 1, type: 'close', data: { reason: "test reason" } }]
)
assert(called, ".disconnect() was not called");
});
});
});
describe("class Connection", function() {
let testConnection = new link.Connection('target', new MockSocket());
it("should send ready after constructor", function() {
assert.deepEqual(
testConnection.socket.sentMessages,
[{seq: 2, type: 'ready', data: {}}]
);
testConnection.socket.sentMessages = [];
});
it("should disconnect when receiving a close message", function() {
testConnection.socket.disconnectCalled = false;
testConnection.socket.events.get('message')({
seq: 1, type: 'close', data: { reason: "test close" }
});
assert(testConnection.socket.disconnectCalled, "Disconnect was not called");
});
it("should close when receiving an invalid message", function() {
testConnection.socket.events.get('message')({ invalid: true });
assert.deepEqual(
testConnection.socket.sentMessages,
[{seq: testConnection._seq - 1, type: 'close', data: { reason: "Invalid message: Malformed message" }}]
);
testConnection.socket.sentMessages = [];
});
describe(".disconnect()", function() {
it("should call disconnect on the socket", function() {
testConnection.socket.disconnectCalled = false;
testConnection.disconnect();
assert(testConnection.socket.disconnectCalled, "Disconnect was not called");
});
});
});
describe("class Client", function() {
let testClient = new link.Client('source', 'url', 'token');
testClient.socket = new MockSocket();
describe(".disconnect()", function() {
it("should call close on the socket", function() {
testClient.socket.closeCalled = false;
testClient.disconnect();
assert(testClient.socket.closeCalled, "Close was not called");
});
});
describe(".register()", function() {
it("is abstract", function() {
assert.throws(
() => testClient.register(),
new Error("Abstract function")
);
});
});
describe(".processHandshake()", function() {
it("should close on invalid message", function() {
testClient._processHandshake({ data: "invalid message" }),
assert.deepEqual(
testClient.socket.sentMessages,
[{seq: testClient._seq - 1, type: 'close', data: {
reason: "Invalid handshake" }
}]
);
testClient.socket.sentMessages = [];
});
it("should call register on hello", function() {
let called = false;
testClient.register = () => { called = true; };
testClient._processHandshake({ seq: 1, type: 'hello', data: { version: "test" }}),
assert(called, "register was not called");
});
it("should emit ready on ready", async function() {
let result = events.once(testClient._events, 'ready');
testClient._processHandshake({ seq: 1, type: 'ready', data: {}}),
await result;
});
it("should emit error on close", async function() {
let result = events.once(testClient._events, 'ready');
testClient._processHandshake({ seq: 1, type: 'close', data: { reason: "test" }}),
await assert.rejects(result, new Error("server closed during handshake: test"));
});
});
describe("._attachSocketHandlers()", function() {
it("should attach handlers", function() {
testClient._attachSocketHandlers();
assert(testClient.socket.events.size > 0, "No handlers were attached");
});
it("should throw on message received in invalid state", function() {
testClient._state = "new";
assert.throws(
() => testClient.socket.events.get('message')(),
new Error("Received message in unexpected state new")
);
});
it("should call _processHandshake on message in handshake state", function() {
testClient._state = "handshake";
let called = false;
testClient._processHandshake = () => { called = true; };
testClient.socket.events.get('message')();
assert(called, "_processHandshake was not called");
});
it("should call processMessage on message in ready state", function() {
testClient._state = "ready";
let called = false;
testClient.processMessage = () => { called = true; };
testClient.socket.events.get('message')();
assert(called, "processMessage was not called");
delete testClient.processMessage;
});
it("should close on invalid message in ready state", function() {
testClient._state = "ready";
testClient.socket.events.get('message')({ data: "invalid message"});
assert.deepEqual(
testClient.socket.sentMessages,
[{seq: testClient._seq - 1, type: 'close', data: {
reason: "Invalid message: Malformed message" }
}]
);
testClient.socket.sentMessages = [];
});
});
});
});
test/lib/link/messages.js
+151
View File
@@ -0,0 +1,151 @@
const assert = require('assert').strict;
const link = require('lib/link');
const errors = require('lib/errors');
describe("lib/link/messages", function() {
let testSourceLink = new link.Link('source', 'target', null);
let testTargetLink = new link.Link('target', 'source', null);
let lastTargetSent;
testTargetLink.send = (type, data) => { lastTargetSent = { type, data }; };
describe("class Request", function() {
let testRequest = new link.Request({
type: 'test',
sources: 'source',
targets: 'target',
});
describe(".attach()", function() {
it("should attach validator to a source link", function() {
testRequest.attach(testSourceLink);
assert(testSourceLink._validators.has('test_response'), "Validator was not set");
});
it("should throw if missing handler on target link", function() {
assert.throws(
() => testRequest.attach(testTargetLink),
new Error("Missing handler for test_request on source-target link")
);
});
let handlerResult;
it("should attach handler to a target link", function() {
testRequest.attach(testTargetLink, async (message) => handlerResult );
assert(testTargetLink._handlers.has('test_request'), "Handler was not set");
});
it("should throw on handler returning object with seq", async function() {
handlerResult = { seq: 3 };
testTargetLink._handlers.get('test_request')({ seq: 2 });
let result = await new Promise(resolve => {
testTargetLink.send = (type, data) => resolve({ type, data });
});
assert.deepEqual(result, {
type: 'test_response',
data: {
error: "response contains reserved property 'seq'",
seq: 2,
},
});
});
it("should send result of calling the handler", async function() {
handlerResult = { test: "handler" };
testTargetLink._handlers.get('test_request')({ seq: 2 });
let result = await new Promise(resolve => {
testTargetLink.send = (type, data) => resolve({ type, data });
});
assert.deepEqual(result, {
type: 'test_response',
data: {
test: "handler",
seq: 2,
},
});
});
it("should implicitly send an empty object on empty return", async function() {
handlerResult = undefined;
testTargetLink._handlers.get('test_request')({ seq: 2 });
let result = await new Promise(resolve => {
testTargetLink.send = (type, data) => resolve({ type, data });
});
assert.deepEqual(result, {
type: 'test_response',
data: {
seq: 2,
},
});
});
});
describe(".send()", function() {
let request;
testSourceLink.send = (type, data) => {
request = { type, data };
};
it("should send request with send and use waitFor to get response", async function() {
testSourceLink.waitFor = (type, condition) => {
return { data: { type, request }};
};
assert.deepEqual(
await testRequest.send(testSourceLink, { test: "request" }),
{ type: 'test_response', request: { type: 'test_request', data: {test: "request" }}}
);
});
it("should throw error response", async function() {
testSourceLink.waitFor = (type, condition) => {
return { data: { error: "test error" }};
};
assert.rejects(
testRequest.send(testSourceLink),
new errors.RequestError("test error")
);
});
});
});
describe("class InstanceRequest", function() {
});
describe("class Event", function() {
let testEvent = new link.Event({
type: 'test',
sources: 'source',
targets: 'target',
});
describe(".attach()", function() {
it("should throw if missing handler on target link", function() {
assert.throws(
() => testEvent.attach(testTargetLink),
new Error("Missing handler for test_event on source-target link")
);
});
let called = false;
it("should attach handler and validator to a target link", function() {
testEvent.attach(testTargetLink, async (message) => { called = true; });
assert(testTargetLink._validators.has('test_event'), "Validator was not set");
});
it("should call the attached event handler", function() {
testTargetLink._handlers.get("test_event")();
assert(called, "Handler was not called");
});
});
describe(".send()", function() {
it("should send the event over the link", async function() {
let sentData = [];
testSourceLink.send = (type, data) => { sentData.push({ type, data }); };
testEvent.send(testSourceLink, { test: "event" });
assert.deepEqual(sentData, [{ type: "test_event", data: { test: "event" }}]);
});
});
});
describe("attachAllMessages()", function() {
it("does not throw", function() {
link.attachAllMessages(testSourceLink);
})
});
});
test/missing.js
+1 -2
View File
@@ -1,10 +1,9 @@
// These are libraries that does not have any units test, they are
// required here in order to track coverage for them.
require('lib/clusterTools');
require('lib/getInstances');
require('lib/manager/instanceManager');
require('lib/manager/modManager');
require('lib/manager/pluginManager');
require('lib/npmPostinstall.js');
require('lib/typedef');
require('../clusterctl.js');