"use strict"; const events = require("events"); const WebSocket = require("ws"); const schema = require("lib/schema"); const errors = require("lib/errors"); class WebSocketBaseConnector extends events.EventEmitter { constructor() { super(); this._socket = null; this._seq = 1; this._lastHeartbeat = null; this._heartbeatId = null; this._heartbeatInterval = null; this._lastReceivedSeq = null; this._sendBuffer = []; } _dropSendBufferSeq(seq) { if (seq === null) { return; } let dropCount = 0; for (let index = 0; index < this._sendBuffer.length; index++) { if (this._sendBuffer[index].seq <= seq) { dropCount++; } else { break; } } this._sendBuffer.splice(0, dropCount); } _doHeartbeat() { this._check(["connected"]); if (Date.now() - this._lastHeartbeat > 2000 * this._heartbeatInterval) { console.log("SOCKET | closing after heartbeat timed out"); this._socket.close(1008, "Heartbeat timeout"); clearInterval(this._heartbeatId); this._heartbeatId = null; } else { this._socket.send(JSON.stringify({ seq: null, type: "heartbeat", data: { seq: this._lastReceivedSeq }, })); } } _processHeartbeat(message) { if (!schema.heartbeat(message)) { console.log("SOCKET | closing after received invalid heartbeat:", message); this._socket.close(1002, "Invalid heartbeat"); console.log(schema.heartbeat.errors); return; } this._lastHeartbeat = Date.now(); this._dropSendBufferSeq(message.data.seq); } /** * Start sending heartbeats */ startHeartbeat() { this._lastHeartbeat = Date.now(); this._heartbeatId = setInterval(() => { this._doHeartbeat(); }, this._heartbeatInterval * 1000); } /** * Stop sending heartbeats */ stopHeartbeat() { if (this._heartbeatId) { clearInterval(this._heartbeatId); this._heartbeatId = null; } } /** * Send a message over the socket * * @returns the sequence number of the message sent */ send(type, data) { this._check(["handshake", "connected", "closing"]); let message = { seq: this._seq, type, data }; this._sendBuffer.push(message); if (["connected", "closing"].includes(this._state)) { this._socket.send(JSON.stringify(message)); } return this._seq++; } /** * Set connection state to closing * * Signal the connection is about tho close and that once the close * frame is received it should be considered finished. */ setClosing() { this._state = "closing"; } } /** * Connector for master server clients */ class WebSocketClientConnector extends WebSocketBaseConnector { constructor(url, reconnectDelay) { super(); this._url = url; // The following states are used in the client connector // new: Not connected // handshake: Attempting to (re)connect to server. // connected: Connection is online // closing: Connection is in the process of being closed. this._state = "new"; this._reconnectId = null; this._sessionToken = null; this._connected = false; this._reconnectDelay = reconnectDelay; } _check(expectedStates) { if (!expectedStates.includes(this._state)) { throw new Error( `Expected state ${expectedStates} but state is ${this._state}` ); } } /** * Send a handshake message over the socket * * Used in the register function in order to send the register handshake * message over the WebSocket. */ sendHandshake(type, data) { this._check(["handshake"]); this._socket.send(JSON.stringify({ seq: null, type, data })); } /** * Close the connection with the given reason. * * Sends a close frame and disconnects the connector. */ close(code, reason) { if (this._state === "new") { return; } if (this._reconnectId) { clearTimeout(this._reconnectId); this._reconnectId = null; } this.stopHeartbeat(); this._state = "closing"; this._socket.close(code, reason); } /** * Connect to the master server */ async connect() { this._check(["new"]); this._state = "handshake"; this._doConnect(); // Wait for the connection to be ready await events.once(this, "connected"); } async _doConnect() { let url = new URL(this._url); url.pathname += "api/socket"; // Open WebSocket to master console.log(`SOCKET | connecting to ${url}`) this._socket = new WebSocket(url, { // For now we do not verify TLS certificates since the default setup is // to create a self-signed certificate. rejectUnauthorized: false, }); this._attachSocketHandlers(); } /** * Attempt re-establishing the connecting again */ reconnect() { if (this._reconnectId !== null) { return; } let delay = Math.random() * this._reconnectDelay; console.log(`SOCKET | waiting ${delay.toLocaleString("en", { maximumFractionDigits: 2 })} seconds for reconnect`) this._reconnectId = setTimeout(() => { this._reconnectId = null; this._doConnect(); }, delay * 1000); } _attachSocketHandlers() { this._socket.on("close", (code, reason) => { console.log(`SOCKET | Close (code: ${code}, reason: ${reason})`); // Authentication failed if (code === 4003) { this.emit("error", new errors.AuthenticationFailed("Authentication failed")); this._state = "closing"; } if (this._state === "closing") { this._lastReceivedSeq = null; this._sessionToken = null; this._sendBuffer.length = 0; this._state = "new"; if (this._connected) { this.emit("close"); this._connected = false; } } else { this._state = "handshake"; this.reconnect(); if (this._connected) { this.emit("dropped"); this._connected = false; } } this.stopHeartbeat(); }); this._socket.on("error", err => { // It's assumed that close is always called by ws console.error("SOCKET | Error:", err); }); /* this._socket.on("unexpected-response", (req, res) => { let data = ""; res.on('data', (chunk) => { data += chunk; }); res.on('end', () => { console.error(`SOCKET | Unexpeced Response ${res.statusCode}`); console.error(data); this._socket.emit("error", new Error(`Unexpected Response ${res.statusCode}`)); }); }); */ this._socket.on("open", () => { console.log("SOCKET | Open"); }); this._socket.on("ping", data => { console.log(`SOCKET | Ping (data: ${data}`); }); this._socket.on("pong", data => { console.log(`SOCKET | Pong (data: ${data}`); }); // Handle messages this._socket.on("message", data => { let message = JSON.parse(data); if (this._state === "handshake") { this._processHandshake(message); } else if (["connected", "closing"].includes(this._state)) { if (message.seq !== null) { this._lastReceivedSeq = message.seq; } if (message.type === "heartbeat") { this._processHeartbeat(message); } else { this.emit("message", message); } } else { throw new Error(`Received message in unexpected state ${this._state}`); } }); } _processHandshake(message) { if (!schema.serverHandshake(message)) { console.log("SOCKET | closing after received invalid handshake:", message); this._socket.close(1002, "Invalid handshake"); return; } let { seq, type, data } = message; if (type == 'hello') { console.log(`SOCKET | received hello from master version ${data.version}`); if (this._sessionToken) { console.log("SOCKET | Attempting resume"); this.sendHandshake("resume", { session_token: this._sessionToken, last_seq: this._lastReceivedSeq, }); } else { this.register(); } } else if (type === 'ready') { console.log("SOCKET | received ready from master"); this._state = "connected"; this._sessionToken = data.session_token; this._heartbeatInterval = data.heartbeat_interval; this.startHeartbeat(); this._connected = true; this.emit("connected"); } else if (type === "continue") { console.log("SOCKET | resuming existing session"); this._state = "connected"; this._heartbeatInterval = data.heartbeat_interval; this.startHeartbeat(); this._dropSendBufferSeq(data.last_seq); for (let message of this._sendBuffer) { this._socket.send(JSON.stringify(message)); } this._connected = true; this.emit("connected"); } else if (type === "invalidate") { this._heartbeatInterval = null; this._lastReceivedSeq = null; this._sessionToken = null; this._sendBuffer.length = 0; this.emit("invalidate"); this.register(); } } /** * 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"); } } class VirtualConnector extends events.EventEmitter { constructor() { super(); this.other = this; this._seq = 1; } /** * Create a pair of virtual connector hooked into each other * * Creates two virtual connectors that are liked to each othes such that * a message sent on one is received by the other. * * @returns {Array} two virtuarl connectors. */ static makePair() { let first = new this(); let second = new this(); first.other = second; second.other = first; return [first, second]; } /** * Send a message to the other end of the connector * * @returns the sequence number of the message sent */ send(type, data) { this.other.emit('message', { seq: this._seq, type, data }); return this._seq++; } } module.exports = { WebSocketBaseConnector, WebSocketClientConnector, VirtualConnector, };