From 3bccbac496d820f09325307da031aefe9f8f1cf5 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Fri, 29 Nov 2019 16:49:34 +0100 Subject: [PATCH] Refactor to a connector based concept in lib/link Remove connection and client subclasses of Link and put the connection logic into a "connector" class. The connector deals with sending and receiving messages, handling the connection and reconnection of lost connections and giving sequence numbers to messages. This move the logic of the actual connection away from the link sub-classes, which in turn means the connection should be a lot easier to mock in tests. --- client.js | 38 ++++-- clusterctl.js | 36 +++--- lib/link/connectors.js | 175 +++++++++++++++++++++++++++ lib/link/index.js | 1 + lib/link/link.js | 241 +++---------------------------------- lib/link/messages.js | 8 +- master.js | 103 ++++++++-------- test/lib/link/connector.js | 108 +++++++++++++++++ test/lib/link/link.js | 181 +--------------------------- test/lib/link/messages.js | 32 +++-- test/master.js | 12 ++ test/mock.js | 51 ++++++++ 12 files changed, 492 insertions(+), 494 deletions(-) create mode 100644 lib/link/connectors.js create mode 100644 test/lib/link/connector.js create mode 100644 test/mock.js 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, +};