factorio-clusterio/test/lib/link/connector.js

import assert from "node:assert/strict";
import events from "node:events";

import * as lib from "@clusterio/lib";
import * as mock from "../../mock.js";
import "../../setup_logging.js";


describe("lib/link/connectors", function() {
	describe("class WebSocketBaseConnector", function() {
		let testConnector = new lib.WebSocketBaseConnector();

		describe("._dropSendBufferSeq()", function() {
			beforeEach(function() {
				testConnector._sendBuffer = [
					{ seq: 1, text: "a", size: 1 },
					{ seq: 2, text: "bb", size: 2 },
					{ seq: 4, text: "cccc", size: 4 },
				];
				testConnector._sendBufferSize = 7;
			});
			it("should do nothing if passed null", function() {
				testConnector._dropSendBufferSeq(null);
				assert(testConnector._sendBuffer.length === 3, "entries were dropped");
			});
			it("should drop up to sequence", function() {
				testConnector._dropSendBufferSeq(2);
				assert.equal(testConnector._sendBuffer.length, 1, "incorrect number of entries dropped");
				assert.equal(testConnector._sendBufferSize, 4);
			});
			it("should ignore holes", function() {
				testConnector._dropSendBufferSeq(3);
				assert.equal(testConnector._sendBuffer.length, 1, "incorrect number of entries dropped");
			});
		});
	});

	describe("class WebSocketClientConnector", function() {
		let testConnector = new lib.WebSocketClientConnector("url", 1);
		testConnector._socket = new mock.MockSocket();

		describe(".sendHandshake()", function() {
			it("calls send on the socket", function() {
				testConnector._state = "connecting";
				testConnector._socket.sentMessages = [];
				testConnector.sendHandshake(new lib.MessageInvalidate());
				assert.deepEqual(
					testConnector._socket.sentMessages.map(JSON.parse),
					[{ type: "invalidate" }]
				);
			});
		});

		describe(".send()", function() {
			it("calls send on the socket", function() {
				testConnector._state = "connected";
				testConnector._socket.sentMessages = [];
				testConnector.send(new lib.MessageHeartbeat(1));
				assert.deepEqual(
					testConnector._socket.sentMessages.map(JSON.parse),
					[{ seq: 1, type: "heartbeat"}]
				);
			});
		});

		describe("send buffer limit", function() {
			const addr = new lib.Address(lib.Address.control, 1);
			function textEvent(seq, text) {
				return new lib.MessageEvent(seq, addr, addr, "TestEvent", { text });
			}
			function utf8Size(message) {
				return Buffer.byteLength(JSON.stringify(message), "utf8");
			}
			function clearBuffer(connector) {
				connector._state = "connected";
				connector._sendBuffer = [];
				connector._sendBufferSize = 0;
				connector._socket.closeCalled = false;
				connector._socket.sentMessages = [];
			}

			afterEach(function() {
				testConnector.maxSendBufferSize = 0;
				testConnector._closing = false;
			});
			it("should count buffered messages in UTF-8 bytes", function() {
				// Extra bytes over the UTF-16 length for ASCII, 2 and 3 byte BMP characters and a surrogate pair
				for (const [text, extraBytes] of [["ascii", 0], ["\u00e9", 1], ["\u20ac", 2], ["\u{1f600}", 2]]) {
					clearBuffer(testConnector);
					const event = textEvent(1, text);
					testConnector.send(event);
					const expected = JSON.stringify(event).length + extraBytes;
					assert.equal(utf8Size(event), expected, text);
					assert.equal(testConnector._sendBufferSize, expected, text);
				}
			});
			it("should allow the limit to be reached exactly and close one byte over it", function() {
				const event = textEvent(1, "\u20ac\u20ac\u20ac\u20ac");
				clearBuffer(testConnector);
				testConnector.maxSendBufferSize = utf8Size(event);
				testConnector.send(event);
				assert.equal(testConnector._sendBuffer.length, 1);
				assert(!testConnector._socket.closeCalled, "Close was called on the socket");

				clearBuffer(testConnector);
				testConnector.maxSendBufferSize = utf8Size(event) - 1;
				assert(JSON.stringify(event).length <= testConnector.maxSendBufferSize);
				testConnector.send(event);
				assert.equal(testConnector._sendBuffer.length, 0);
				assert.equal(testConnector._sendBufferSize, 0);
				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
				assert.equal(testConnector._socket.sentMessages.length, 0);
			});
			it("should subtract the bytes of acknowledged messages", function() {
				const messages = [textEvent(1, "\u00e9"), textEvent(2, "\u20ac"), textEvent(3, "\u{1f600}")];
				clearBuffer(testConnector);
				for (const event of messages) {
					testConnector.send(event);
				}
				assert.equal(testConnector._sendBufferSize, messages.reduce((sum, event) => sum + utf8Size(event), 0));
				testConnector._dropSendBufferSeq(2);
				assert.equal(testConnector._sendBuffer.length, 1);
				assert.equal(testConnector._sendBufferSize, utf8Size(messages[2]));
				testConnector._dropSendBufferSeq(3);
				assert.equal(testConnector._sendBuffer.length, 0);
				assert.equal(testConnector._sendBufferSize, 0);
			});
			it("should clear the byte count when the session is invalidated or reset", function() {
				const connector = new lib.WebSocketClientConnector("url", 1);
				connector._socket = new mock.MockSocket();
				clearBuffer(connector);
				connector.send(textEvent(1, "\u20ac"));
				connector._invalidate();
				assert.equal(connector._sendBuffer.length, 0);
				assert.equal(connector._sendBufferSize, 0);

				clearBuffer(connector);
				connector.send(textEvent(1, "\u20ac"));
				connector._socket = null;
				connector._reset();
				assert.equal(connector._sendBuffer.length, 0);
				assert.equal(connector._sendBufferSize, 0);
			});
			it("should keep unacknowledged messages under the limit", function() {
				testConnector._state = "connected";
				testConnector._sendBuffer = [];
				testConnector._sendBufferSize = 0;
				testConnector._socket.closeCalled = false;
				testConnector.maxSendBufferSize = 1000;
				testConnector.send(new lib.MessageHeartbeat(1));
				assert.equal(testConnector._sendBuffer.length, 1);
				assert(!testConnector._socket.closeCalled, "Close was called on the socket");
			});
			it("should clear the buffer and close when the limit is exceeded", function() {
				testConnector._state = "connected";
				testConnector._sendBuffer = [];
				testConnector._sendBufferSize = 0;
				testConnector._socket.closeCalled = false;
				testConnector._socket.sentMessages = [];
				testConnector.maxSendBufferSize = 30;
				testConnector.send(new lib.MessageHeartbeat(1));
				testConnector.send(new lib.MessageHeartbeat(2));
				assert.equal(testConnector._sendBuffer.length, 0);
				assert.equal(testConnector._sendBufferSize, 0);
				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
				assert.equal(testConnector._socket.sentMessages.length, 1);
			});
		});

		describe(".close()", function() {
			it("should call close", function() {
				testConnector._socket.closeCalled = false;
				testConnector.close("test reason");
				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
			});
		});

		describe(".parseMessage()", function() {
			it("should close on invalid json", function() {
				testConnector._socket.closeCalled = false;
				testConnector._parseMessage("invalid json");
				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
			});
			it("should close on invalid message", function() {
				testConnector._socket.closeCalled = false;
				testConnector._parseMessage('{"invalid":"message"}');
				assert(testConnector._socket.closeCalled, "Close was not called on the socket");
			});
		});
		describe(".processHandshake()", function() {
			it("should call register on hello", function() {
				let called = false;
				testConnector.register = () => { called = true; };
				testConnector._processHandshake(
					new lib.MessageHello(
						new lib.HelloData("test", {})
					)
				);
				assert(called, "register was not called");
			});
			it("should emit connect on ready", async function() {
				let result = events.once(testConnector, "connect");
				testConnector._processHandshake(
					new lib.MessageReady(
						new lib.ReadyData(
							new lib.Address(lib.Address.control, 1),
							"x",
							20,
							10,
							undefined
						)
					)
				);
				await result;
				clearInterval(testConnector._heartbeatId);
			});
		});

		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 = "closed";
				assert.throws(
					() => testConnector._socket.events.get("message")(JSON.stringify(new lib.MessageInvalidate())),
					new Error("Received message in unexpected state closed")
				);
			});
			it("should call _processHandshake on message in handshake state", function() {
				testConnector._state = "connecting";
				let called = false;
				testConnector._processHandshake = () => { called = true; };
				testConnector._socket.events.get("message")(JSON.stringify(new lib.MessageInvalidate()));
				assert(called, "_processHandshake was not called");
			});
			it("should emit message on message in connected state", async function() {
				testConnector._socket.sentMessages = [];
				testConnector._state = "connected";
				let result = events.once(testConnector, "message");
				let addr = new lib.Address(lib.Address.control, 1);
				let event = new lib.MessageEvent(1, addr, addr, "TestEvent");
				testConnector._socket.events.get("message")(JSON.stringify(event));
				assert.deepEqual((await result)[0], event);
			});
		});
	});
});