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

Refactor to a connector based concept in lib/link

Browse Source 
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.
This commit is contained in:
Hornwitser committed 2019-11-30 00:08:10 +01:00
1 parent a3eb39712c
commit 3bccbac496
12 files changed
+492 -494
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

client.js
+25 -13
View File
@@ -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();
/*
clusterctl.js
+21 -15
View File
@@ -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);
lib/link/connectors.js
+175
View File
@@ -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,
};
lib/link/index.js
+1
View File
@@ -1,4 +1,5 @@
module.exports = {
...require('lib/link/link'),
...require('lib/link/messages'),
...require('lib/link/connectors'),
};
lib/link/link.js
+20 -221
View File
@@ -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,
}
lib/link/messages.js
+4 -4
View File
@@ -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);
}
}
master.js
+54 -49
View File
@@ -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) {
test/lib/link/connector.js
+108
View File
@@ -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"]);
});
});
});
});
test/lib/link/link.js
+2 -179
View File
@@ -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 = [];
});
});
});
});
test/lib/link/messages.js
+19 -13
View File
@@ -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" }}]
);
});
});
});
test/master.js
+12
View File
@@ -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");
});
});
});
});
test/mock.js
+51
View File
@@ -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,
};