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); }); }); }); });