diff --git a/client.js b/client.js index 2fa2f555..ed2d79a5 100644 --- a/client.js +++ b/client.js @@ -235,17 +235,36 @@ class Instance { } } +class SlaveConnector extends link.SocketIOClientConnector { + constructor(slaveConfig) { + super(slaveConfig.masterURL, slaveConfig.masterAuthToken); + + this.id = slaveConfig.id; + this.name = slaveConfig.name; + } + + register() { + console.log("SOCKET | registering slave"); + this.send('register_slave', { + agent: 'Clusterio Slave', + version, + id: this.id, + name: this.name, + }); + } +} + /** * Handles running the slave * * Connects to the master server over the socket.io connection and manages * intsances. */ -class Slave extends link.Client { +class Slave extends link.Link { // 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); + constructor(connector, slaveConfig) { + super('slave', 'master', connector); link.attachAllMessages(this); this.config = { id: slaveConfig.id, @@ -256,16 +275,7 @@ class Slave extends link.Client { 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) { @@ -626,7 +636,8 @@ async function startClient() { return; } - let slave = new Slave(slaveConfig); + let slaveConnector = new SlaveConnector(slaveConfig); + let slave = new Slave(slaveConnector, slaveConfig); // Handle interrupts let secondSigint = false @@ -645,6 +656,7 @@ async function startClient() { }); }); + await slaveConnector.connect(); await slave.start(); /* diff --git a/clusterctl.js b/clusterctl.js index 29e9772d..ce5693ad 100644 --- a/clusterctl.js +++ b/clusterctl.js @@ -253,20 +253,9 @@ 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. + * Connector for control connection to master server */ -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); - } - +class ControlConnector extends link.SocketIOClientConnector { register() { console.log("SOCKET | registering control"); this.send('register_control', { @@ -274,6 +263,22 @@ class Control extends link.Client { version: version, }); } +} + +/** + * Handles running the control + * + * Connects to the master server over the socket.io connection and sends + * commands to it. + */ +class Control extends link.Link { + + // I don't like God classes, but the alternative of putting all this state + // into global variables is not much better. + constructor(connector) { + super('control', 'master', connector); + link.attachAllMessages(this); + } async instanceOutputEventHandler(message) { let { instance_id, output } = message.data; @@ -395,8 +400,9 @@ async function startControl() { } } - let control = new Control(controlConfig); - await control.connect(); + let controlConnector = new ControlConnector(controlConfig.url, controlConfig.token); + let control = new Control(controlConnector); + await controlConnector.connect(); if (commands.has(commandName)) { command = commands.get(commandName); diff --git a/lib/link/connectors.js b/lib/link/connectors.js new file mode 100644 index 00000000..c124484d --- /dev/null +++ b/lib/link/connectors.js @@ -0,0 +1,175 @@ +"use strict"; +const events = require("events"); +const io = require("socket.io-client"); + +const schema = require("lib/schema"); + + +/** + * Connector for master server clients + */ +class SocketIOClientConnector extends events.EventEmitter { + constructor(url, token) { + super(); + + this._url = url; + this._token = token; + + this._seq = 1; + this._state = "new"; + } + + _check(expectedStates) { + if (!expectedStates.includes(this._state)) { + throw new Error( + `Expected state ${expectedStates} but state is ${this._state}` + ); + } + } + + /** + * Send a message over the socket + * + * @returns the sequence number of the message sent + */ + send(type, data) { + this._socket.send({ seq: this._seq, type, data }); + return this._seq++; + } + + /** + * Close the connection with the given reason. + * + * Sends a close message and disconnects the connector. + */ + close(reason) { + this.send('close', { reason }); + this.disconnect(); + } + + /** + * Connect to the master server + */ + 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, + }); + + this._attachSocketHandlers(); + + // Wait for the connection to be ready + await events.once(this, '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") { + this.emit("message", message); + } else { + throw new Error(`Received message in unexpected state ${this._state}`); + } + }); + } + + /** + * Immediatly close the connection + */ + disconnect() { + this._socket.close(); + } + + _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.emit('ready'); + + } else if (type === 'close') { + console.log(`SOCKET | received close from server: ${data.reason}`); + this.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 = { + SocketIOClientConnector, +}; diff --git a/lib/link/index.js b/lib/link/index.js index c8641a97..132ff5d0 100644 --- a/lib/link/index.js +++ b/lib/link/index.js @@ -1,4 +1,5 @@ module.exports = { ...require('lib/link/link'), ...require('lib/link/messages'), + ...require('lib/link/connectors'), }; diff --git a/lib/link/link.js b/lib/link/link.js index a6398d21..3260b430 100644 --- a/lib/link/link.js +++ b/lib/link/link.js @@ -1,6 +1,4 @@ "use strict"; -const events = require("events"); -const io = require("socket.io-client"); const schema = require("lib/schema"); const errors = require("lib/errors"); @@ -8,6 +6,7 @@ const messages = require("lib/link/messages"); // Some definitions for the terminology used here: // link: Either side of a master - client connection +// connector: Adapter between a link and a socket 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 @@ -18,15 +17,31 @@ const messages = require("lib/link/messages"); * Common interface for server and client connections */ class Link { - constructor(source, target, socket) { + constructor(source, target, connector) { this.source = source; this.target = target; - this.socket = socket; + this.connector = connector; - this._seq = 1; this._waiters = new Map(); this._handlers = new Map(); this._validators = new Map(); + + // Process messages received by the connector + connector.on('message', payload => { + try { + this.processMessage(payload); + } catch (err) { + if (err instanceof errors.InvalidMessage) { + this.connector.close(`Invalid message: ${err.message}`); + } else { + throw err; + } + } + }); + + this.setHandler('close', message => { + this.connector.disconnect(message.data.reason); + }, schema.close); } /** @@ -154,16 +169,6 @@ class Link { 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 * @@ -185,215 +190,9 @@ class Link { } }); } - - /** - * 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, } diff --git a/lib/link/messages.js b/lib/link/messages.js index 8007290e..3f86989e 100644 --- a/lib/link/messages.js +++ b/lib/link/messages.js @@ -96,13 +96,13 @@ class Request { if (Object.prototype.hasOwnProperty.call(response, 'seq')) { throw new errors.RequestError("response contains reserved property 'seq'"); } - link.send(this.responseType, { seq: message.seq, ...response }); + link.connector.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 }); + link.connector.send(this.responseType, { seq: message.seq, error: err.message }); }); }, this._requestValidator); } @@ -118,7 +118,7 @@ class Request { */ async send(link, data = {}) { // XXX validate link target/source? - let seq = link.send(this.requestType, data); + let seq = link.connector.send(this.requestType, data); let responseMessage = await link.waitFor(this.responseType, { seq }); if (responseMessage.data.error) { throw new errors.RequestError(responseMessage.data.error); @@ -388,7 +388,7 @@ class Event { * Sends the given event data over the link. */ send(link, data = {}) { - link.send(this.eventType, data); + link.connector.send(this.eventType, data); } } diff --git a/master.js b/master.js index d312a6f7..5a6e67c9 100644 --- a/master.js +++ b/master.js @@ -24,6 +24,7 @@ const crypto = require('crypto'); const base64url = require('base64url'); const moment = require("moment"); const request = require("request"); +const events = require("events"); const version = require("./package").version; // constants @@ -248,9 +249,9 @@ const ioMetrics = require("socket.io-prometheus"); ioMetrics(io); -class BaseConnection extends link.Connection { - constructor(target, socket) { - super(target, socket); +class BaseConnection extends link.Link { + constructor(target, connector) { + super('master', target, connector); link.attachAllMessages(this); } @@ -271,13 +272,13 @@ class BaseConnection extends link.Connection { let controlConnections = new Array(); class ControlConnection extends BaseConnection { - constructor(registerData, socket) { - super('control', socket) + constructor(registerData, connector) { + super('control', connector) this._agent = registerData.agent; this._version = registerData.version; - socket.on('disconnect', () => { + this.connector.on('disconnect', () => { let index = controlConnections.indexOf(this); if (index !== -1) { controlConnections.splice(index, 1); @@ -344,8 +345,8 @@ class ControlConnection extends BaseConnection { var slaveConnections = new Map(); class SlaveConnection extends BaseConnection { - constructor(registerData, socket) { - super('slave', socket); + constructor(registerData, connector) { + super('slave', connector); this._agent = registerData.agent; this._id = registerData.id; @@ -359,37 +360,14 @@ class SlaveConnection extends BaseConnection { version: this._version, }); - socket.on('message', () => { + this.connector.on('message', () => { prometheusWsUsageCounter.labels('message', "other").inc(); }); - socket.on('disconnect', () => { + this.connector.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); }); } @@ -445,21 +423,53 @@ function isInteger(value) { } +/** + * Connector for master server connections + */ +class SocketIOServerConnector extends events.EventEmitter { + constructor(socket) { + super(); + + this._seq = 1; + this._socket = socket; + + this._socket.on('message', message => { + this.emit('message', message); + }); + this._socket.on('disconnect', () => { + this.emit('disconnect'); + }); + } + + /** + * Send a message over the socket + * + * @returns the sequence number of the message sent + */ + send(type, data = {}) { + this._socket.send({ seq: this._seq, type, data }); + return this._seq++; + } + + disconnect() { + this._socket.disconnect(true); + } +} + io.on('connection', function (socket) { console.log(`SOCKET | new connection from ${socket.handshake.address}`); // Start connection handshake - socket.send({ seq: 1, type: 'hello', data: { version }}); + let connector = new SocketIOServerConnector(socket); + connector.send('hello', { 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); + connector.send('close', { reason: "Invalid handshake" }); + connector.disconnect(true); return; } @@ -472,26 +482,20 @@ io.on('connection', function (socket) { } console.log(`SOCKET | registered slave ${data.id} version ${data.version}`); - slaveConnections.set(data.id, new SlaveConnection(data, socket)); + slaveConnections.set(data.id, new SlaveConnection(data, connector)); + connector.send('ready'); } else if (type === "register_control") { console.log(`SOCKET | registered control from ${socket.handshake.address}`); - controlConnections.push(new ControlConnection(data, socket)); + controlConnections.push(new ControlConnection(data, connector)); + connector.send('ready'); } else if (type === "close") { console.log(`SOCKET | received close from ${socket.handshake.address}: ${data.reason}`); - socket.disconnect(true); + connector.disconnect(); return; } }); - - // 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 @@ -660,6 +664,7 @@ module.exports = { // For testing only _db: db, _config: config, + _SocketIOServerConnector: SocketIOServerConnector, } if (module === require.main) { diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js new file mode 100644 index 00000000..014baa8c --- /dev/null +++ b/test/lib/link/connector.js @@ -0,0 +1,108 @@ +const assert = require('assert').strict; +const events = require('events'); + +const mock = require('../../mock'); +const link = require('lib/link'); + + +describe("lib/link/connectors", function() { + describe("class SocketIOClientConnector", function() { + let testConnector = new link.SocketIOClientConnector('source', 'url', 'token'); + testConnector._socket = new mock.MockSocket(); + describe(".disconnect()", function() { + it("should call close on the socket", function() { + testConnector._socket.closeCalled = false; + testConnector.disconnect(); + assert(testConnector._socket.closeCalled, "Close was not called"); + }); + }); + + describe(".register()", function() { + it("is abstract", function() { + assert.throws( + () => testConnector.register(), + new Error("Abstract function") + ); + }); + }); + + describe(".send()", function() { + it("calls send on the socket", function() { + testConnector._socket.sentMessages = []; + let seq = testConnector.send('test', { test: true }); + assert.deepEqual(testConnector._socket.sentMessages, [{ seq, type: 'test', data: { test: true }}]); + }); + }); + + describe(".close()", function() { + it("should send close and call disconnect", function() { + testConnector._socket.sentMessages = []; + let called = false; + testConnector.disconnect = () => { called = true; } + testConnector.close("test reason"); + assert.deepEqual( + testConnector._socket.sentMessages, + [{ seq: testConnector._seq - 1, type: 'close', data: { reason: "test reason" } }] + ) + assert(called, ".disconnect() was not called"); + }); + }); + + describe(".processHandshake()", function() { + it("should close on invalid message", function() { + testConnector._socket.sentMessages = []; + testConnector._processHandshake({ data: "invalid message" }), + assert.deepEqual( + testConnector._socket.sentMessages, + [{seq: testConnector._seq - 1, type: 'close', data: { + reason: "Invalid handshake" } + }] + ); + }); + it("should call register on hello", function() { + let called = false; + testConnector.register = () => { called = true; }; + testConnector._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(testConnector, 'ready'); + testConnector._processHandshake({ seq: 1, type: 'ready', data: {}}), + await result; + }); + it("should emit error on close", async function() { + let result = events.once(testConnector, 'ready'); + testConnector._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() { + testConnector._attachSocketHandlers(); + assert(testConnector._socket.events.size > 0, "No handlers were attached"); + }); + it("should throw on message received in invalid state", function() { + testConnector._state = "new"; + assert.throws( + () => testConnector._socket.events.get('message')(), + new Error("Received message in unexpected state new") + ); + }); + it("should call _processHandshake on message in handshake state", function() { + testConnector._state = "handshake"; + let called = false; + testConnector._processHandshake = () => { called = true; }; + testConnector._socket.events.get('message')(); + assert(called, "_processHandshake was not called"); + }); + it("should emit message on message in ready state", async function() { + testConnector._socket.sentMessages = []; + testConnector._state = "ready"; + let result = events.once(testConnector, 'message'); + testConnector._socket.events.get('message')("message"); + assert.deepEqual(await result, ["message"]); + }); + }); + }); +}); diff --git a/test/lib/link/link.js b/test/lib/link/link.js index 44f494f1..8ca161fc 100644 --- a/test/lib/link/link.js +++ b/test/lib/link/link.js @@ -4,35 +4,12 @@ const events = require('events'); const link = require("lib/link"); const errors = require("lib/errors"); const schema = require("lib/schema"); +const mock = require("../../mock"); -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); + let testLink = new link.Link('source', 'target', new mock.MockConnector()); testLink.setValidator('simple', schema.compile({ properties: { "data": { @@ -121,23 +98,6 @@ describe("lib/link/link", function() { }); }); - 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 => { @@ -196,142 +156,5 @@ describe("lib/link/link", function() { 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 = []; - }); - }); }); }); diff --git a/test/lib/link/messages.js b/test/lib/link/messages.js index 461c013d..92085044 100644 --- a/test/lib/link/messages.js +++ b/test/lib/link/messages.js @@ -2,11 +2,12 @@ const assert = require('assert').strict; const link = require('lib/link'); const errors = require('lib/errors'); +const mock = require('../../mock'); describe("lib/link/messages", function() { - let testSourceLink = new link.Link('source', 'target', null); - let testTargetLink = new link.Link('target', 'source', null); + let testSourceLink = new link.Link('source', 'target', new mock.MockConnector()); + let testTargetLink = new link.Link('target', 'source', new mock.MockConnector()); let lastTargetSent; testTargetLink.send = (type, data) => { lastTargetSent = { type, data }; }; @@ -39,7 +40,7 @@ describe("lib/link/messages", function() { handlerResult = { seq: 3 }; testTargetLink._handlers.get('test_request')({ seq: 2 }); let result = await new Promise(resolve => { - testTargetLink.send = (type, data) => resolve({ type, data }); + testTargetLink.connector.send = (type, data) => resolve({ type, data }); }); assert.deepEqual(result, { type: 'test_response', @@ -53,7 +54,7 @@ describe("lib/link/messages", function() { handlerResult = { test: "handler" }; testTargetLink._handlers.get('test_request')({ seq: 2 }); let result = await new Promise(resolve => { - testTargetLink.send = (type, data) => resolve({ type, data }); + testTargetLink.connector.send = (type, data) => resolve({ type, data }); }); assert.deepEqual(result, { type: 'test_response', @@ -67,7 +68,7 @@ describe("lib/link/messages", function() { handlerResult = undefined; testTargetLink._handlers.get('test_request')({ seq: 2 }); let result = await new Promise(resolve => { - testTargetLink.send = (type, data) => resolve({ type, data }); + testTargetLink.connector.send = (type, data) => resolve({ type, data }); }); assert.deepEqual(result, { type: 'test_response', @@ -79,11 +80,11 @@ describe("lib/link/messages", function() { }); 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() { + let request; + testSourceLink.connector.send = (type, data) => { + request = { type, data }; + }; testSourceLink.waitFor = (type, condition) => { return { data: { type, request }}; }; @@ -91,6 +92,8 @@ describe("lib/link/messages", function() { await testRequest.send(testSourceLink, { test: "request" }), { type: 'test_response', request: { type: 'test_request', data: {test: "request" }}} ); + delete testSourceLink.connector.send; + delete testSourceLink.waitFor; }); it("should throw error response", async function() { testSourceLink.waitFor = (type, condition) => { @@ -100,6 +103,7 @@ describe("lib/link/messages", function() { testRequest.send(testSourceLink), new errors.RequestError("test error") ); + delete testSourceLink.waitFor; }); }); }); @@ -134,11 +138,13 @@ describe("lib/link/messages", function() { describe(".send()", function() { it("should send the event over the link", async function() { - let sentData = []; - testSourceLink.send = (type, data) => { sentData.push({ type, data }); }; - + testSourceLink.connector.sentMessages = []; testEvent.send(testSourceLink, { test: "event" }); - assert.deepEqual(sentData, [{ type: "test_event", data: { test: "event" }}]); + let seq = testSourceLink.connector._seq - 1; + assert.deepEqual( + testSourceLink.connector.sentMessages, + [{ seq, type: "test_event", data: { test: "event" }}] + ); }); }); }); diff --git a/test/master.js b/test/master.js index d4dd484d..a4f5a283 100644 --- a/test/master.js +++ b/test/master.js @@ -3,6 +3,7 @@ const request = require('supertest'); const validateHTML = require('html5-validator'); const parallel = require('mocha.parallel'); +const mock = require('./mock'); const master = require('../master'); @@ -43,4 +44,15 @@ describe('Master testing', function() { it(`sends some HTML when accessing ${path}`, () => getValidate(path)); } }); + + describe("class SocketIOServerConnector", function() { + let testConnector = new master._SocketIOServerConnector(new mock.MockSocket()); + describe(".disconnect()", function() { + it("should call disconnect on the socket", function() { + testConnector._socket.disconnectCalled = false; + testConnector.disconnect(); + assert(testConnector._socket.disconnectCalled, "Disconnect was not called"); + }); + }); + }); }); diff --git a/test/mock.js b/test/mock.js new file mode 100644 index 00000000..8c230e44 --- /dev/null +++ b/test/mock.js @@ -0,0 +1,51 @@ +const events = require("events"); + + +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; + } +}; + +class MockConnector extends events.EventEmitter { + constructor() { + super(); + + this._seq = 1; + this.sentMessages = []; + this.events = new Map(); + this.handshake = { address: "socket.test" }; + } + + send(type, data) { + this.sentMessages.push({ seq: this._seq, type, data }); + return this._seq++; + } + + disconnect() { + this.disconnectCalled = true; + } +}; + +module.exports = { + MockSocket, + MockConnector, +};