diff --git a/client/base_client.ts b/client/base_client.ts index 38c2267..0712f13 100644 --- a/client/base_client.ts +++ b/client/base_client.ts @@ -1,5 +1,5 @@ import type { QoS } from "../lib/mod.ts"; -import { encode as connectEncoder } from "../packets/connect.ts"; +import { ConnectPacket, encode as connectEncoder } from "../packets/connect.ts"; import { encode as disconnectEncoder } from "../packets/disconnect.ts"; import type { AnyPacket, @@ -46,6 +46,7 @@ export type ClientOptions = { reconnect?: boolean | RetryOptions; incomingStore?: IncomingStore; outgoingStore?: OutgoingStore; + will?: ConnectPacket["will"]; logger?: (msg: string, ...args: unknown[]) => void; }; @@ -273,15 +274,14 @@ export abstract class Client { protected constructor(options?: ClientOptions) { this.options = options || {}; this.clientId = this.generateClientId(); - this.keepAlive = - typeof this.options.keepAlive === "number" - ? this.options.keepAlive - : defaultKeepAlive; + this.keepAlive = typeof this.options.keepAlive === "number" + ? this.options.keepAlive + : defaultKeepAlive; - this.incomingStore = - this.options.incomingStore || new IncomingMemoryStore(); - this.outgoingStore = - this.options.outgoingStore || new OutgoingMemoryStore(); + this.incomingStore = this.options.incomingStore || + new IncomingMemoryStore(); + this.outgoingStore = this.options.outgoingStore || + new OutgoingMemoryStore(); this.log = this.options.logger || (() => {}); @@ -296,7 +296,9 @@ export abstract class Client { break; default: return Promise.reject( - new Error(`should not be connecting in ${this.connectionState} state`) + new Error( + `should not be connecting in ${this.connectionState} state`, + ), ); } @@ -314,7 +316,7 @@ export abstract class Client { public publish( topic: string, payload: PublishPayload, - options?: PublishOptions + options?: PublishOptions, ): Promise { const dup = (options && options.dup) || false; const qos = (options && options.qos) || 0; @@ -379,33 +381,33 @@ export abstract class Client { public async subscribe( topicFilter: string, - qos?: QoS + qos?: QoS, ): Promise; public async subscribe( topicFilters: string[], - qos?: QoS + qos?: QoS, ): Promise; public async subscribe( subscription: SubscriptionOption, - qos?: QoS + qos?: QoS, ): Promise; public async subscribe( subscriptions: SubscriptionOption[], - qos?: QoS + qos?: QoS, ): Promise; public async subscribe( input: SubscriptionOption | string | (SubscriptionOption | string)[], - qos?: QoS + qos?: QoS, ): Promise { switch (this.connectionState) { case "disconnecting": case "disconnected": throw new Error( - `should not be subscribing in ${this.connectionState} state` + `should not be subscribing in ${this.connectionState} state`, ); } @@ -413,10 +415,10 @@ export abstract class Client { const subs = arr.map((sub) => { return typeof sub === "object" ? { - topicFilter: sub.topicFilter, - qos: sub.qos || qos || 0, - state: "pending", - } + topicFilter: sub.topicFilter, + qos: sub.qos || qos || 0, + state: "pending", + } : { topicFilter: sub, qos: qos || 0, state: "pending" }; }); const promises = []; @@ -427,7 +429,7 @@ export abstract class Client { // to do when it receives a subscribe packet containing a topic filter // matching an existing subscription. this.subscriptions = this.subscriptions.filter( - (old) => old.topicFilter !== sub.topicFilter + (old) => old.topicFilter !== sub.topicFilter, ); this.subscriptions.push(sub); @@ -482,7 +484,7 @@ export abstract class Client { case "disconnecting": case "disconnected": throw new Error( - `should not be unsubscribing in ${this.connectionState} state` + `should not be unsubscribing in ${this.connectionState} state`, ); } @@ -491,7 +493,7 @@ export abstract class Client { for (const topicFilter of arr) { const sub = this.subscriptions.find( - (sub) => sub.topicFilter === topicFilter + (sub) => sub.topicFilter === topicFilter, ) || { topicFilter, qos: 0, state: "unknown" }; const deferred = new Deferred(); const promise = deferred.promise.then(() => sub); @@ -539,7 +541,7 @@ export abstract class Client { for (const sub of this.subscriptions) { if (sub.state === "removed") { const unresolvedSubscribe = this.unresolvedSubscribes.get( - sub.topicFilter + sub.topicFilter, ); if (unresolvedSubscribe) { @@ -549,7 +551,7 @@ export abstract class Client { } const unresolvedUnsubscribe = this.unresolvedUnsubscribes.get( - sub.topicFilter + sub.topicFilter, ); if (unresolvedUnsubscribe) { @@ -565,7 +567,7 @@ export abstract class Client { } this.subscriptions = this.subscriptions.filter( - (sub) => sub.state !== "removed" + (sub) => sub.state !== "removed", ); if (subs.length > 0 && this.connectionState === "connected") { @@ -605,7 +607,7 @@ export abstract class Client { break; default: throw new Error( - `should not be disconnecting in ${this.connectionState} state` + `should not be disconnecting in ${this.connectionState} state`, ); } } @@ -651,9 +653,10 @@ export abstract class Client { username: this.options.username, password: this.options.password, clean: this.options.clean !== false, + will: this.options.will, keepAlive: this.keepAlive, }, - connectEncoder + connectEncoder, ); this.startConnectTimer(); @@ -803,7 +806,7 @@ export abstract class Client { break; default: throw new Error( - `should not be receiving connack packets in ${this.connectionState} state` + `should not be receiving connack packets in ${this.connectionState} state`, ); } @@ -822,7 +825,7 @@ export abstract class Client { } else if (packet.qos === 1) { if (typeof packet.id !== "number" || packet.id < 1) { return this.protocolViolation( - "publish packet with qos 1 is missing id" + "publish packet with qos 1 is missing id", ); } @@ -833,17 +836,17 @@ export abstract class Client { type: "puback", id: packet.id, }, - pubackEncoder + pubackEncoder, ); } else if (packet.qos === 2) { if (typeof packet.id !== "number" || packet.id < 1) { return this.protocolViolation( - "publish packet with qos 2 is missing id" + "publish packet with qos 2 is missing id", ); } - const emitMessage = - !packet.dup || !(await this.incomingStore.has(packet.id)); + const emitMessage = !packet.dup || + !(await this.incomingStore.has(packet.id)); if (emitMessage) { this.incomingStore.store(packet.id); @@ -856,7 +859,7 @@ export abstract class Client { type: "pubrec", id: packet.id, }, - pubrecEncoder + pubrecEncoder, ); } } @@ -893,7 +896,7 @@ export abstract class Client { type: "pubcomp", id: packet.id, }, - pubcompEncoder + pubcompEncoder, ); } @@ -912,7 +915,7 @@ export abstract class Client { protected handleSuback(packet: SubackPacket) { const unacknowledgedSubscribe = this.unacknowledgedSubscribes.get( - packet.id + packet.id, ); // TODO: verify returnCodes length matches subscriptions.length @@ -936,14 +939,14 @@ export abstract class Client { } } else { throw new Error( - `received suback packet with unrecognized id ${packet.id}` + `received suback packet with unrecognized id ${packet.id}`, ); } } protected handleUnsuback(packet: UnsubackPacket) { const unacknowledgedUnsubscribe = this.unacknowledgedUnsubscribes.get( - packet.id + packet.id, ); if (unacknowledgedUnsubscribe) { @@ -968,7 +971,7 @@ export abstract class Client { } } else { throw new Error( - `received unsuback packet with unrecognized id ${packet.id}` + `received unsuback packet with unrecognized id ${packet.id}`, ); } } @@ -979,7 +982,7 @@ export abstract class Client { () => { this.connectTimedOut(); }, - this.options.connectTimeout || defaultConnectTimeout + this.options.connectTimeout || defaultConnectTimeout, ); } @@ -989,7 +992,7 @@ export abstract class Client { break; default: throw new Error( - `connect timer should not be timing out in ${this.connectionState} state` + `connect timer should not be timing out in ${this.connectionState} state`, ); } @@ -1032,6 +1035,7 @@ export abstract class Client { defaultOptions = defaultReconnectOptions; } + //@ts-expect-error Cannot be false if (reconnectOptions === false) { return; } else if (reconnectOptions === true) { @@ -1073,7 +1077,7 @@ export abstract class Client { this.reconnectAttempt++; this.openConnection(); }, - delay + delay, ); return true; @@ -1114,7 +1118,7 @@ export abstract class Client { { type: "pingreq", }, - pingreqEncoder + pingreqEncoder, ); // TODO: need a timer here to disconnect if we don't receive the pingres @@ -1135,7 +1139,7 @@ export abstract class Client { protected startTimer( name: string, cb: (...args: unknown[]) => void, - delay: number + delay: number, ) { if (this.timerExists(name)) { this.log(`timer ${name} already exists`); @@ -1208,10 +1212,9 @@ export abstract class Client { } private getURL(): URL { - let url: URL | string | void = - typeof this.options.url === "function" - ? this.options.url() - : this.options.url; + let url: URL | string | void = typeof this.options.url === "function" + ? this.options.url() + : this.options.url; if (!url) { url = this.getDefaultURL(); @@ -1261,7 +1264,7 @@ export abstract class Client { protected async send( packet: T, - encoder: PacketEncoder + encoder: PacketEncoder, ) { this.log(`sending ${packet.type} packet`, packet); @@ -1293,7 +1296,7 @@ export abstract class Client { if (listeners) { this.eventListeners.set( eventName, - listeners.filter((l) => l !== listener) + listeners.filter((l) => l !== listener), ); } } diff --git a/client/buffer_test.ts b/client/buffer_test.ts index 0c5331f..c96c41b 100644 --- a/client/buffer_test.ts +++ b/client/buffer_test.ts @@ -23,7 +23,7 @@ Deno.test("client can receive one byte at a time", async () => { returnCode: 0, sessionPresent: false, }), - { trickle: true } + { trickle: true }, ); assertEquals(client.receivedPackets[0].type, "connack"); @@ -40,9 +40,9 @@ Deno.test("client can receive one byte at a time", async () => { qos: 0, id: 0, }, - new TextEncoder() + new TextEncoder(), ), - { trickle: true } + { trickle: true }, ); assertEquals(client.receivedPackets[1].type, "publish"); @@ -57,7 +57,7 @@ Deno.test("client can receive one byte at a time", async () => { qos: 0, id: 0, }, - new TextEncoder() + new TextEncoder(), ); // Receive all but the last byte: @@ -89,7 +89,7 @@ Deno.test("client can receive bytes for multiple packets at once", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.receivedPackets[0].type, "connack"); @@ -106,7 +106,7 @@ Deno.test("client can receive bytes for multiple packets at once", async () => { qos: 0, id: 0, }, - new TextEncoder() + new TextEncoder(), ), ...encodePublish( { @@ -118,7 +118,7 @@ Deno.test("client can receive bytes for multiple packets at once", async () => { qos: 0, id: 0, }, - new TextEncoder() + new TextEncoder(), ), ]); diff --git a/client/connect_test.ts b/client/connect_test.ts index ad089aa..ebaa27d 100644 --- a/client/connect_test.ts +++ b/client/connect_test.ts @@ -20,7 +20,7 @@ Deno.test("connect/disconnect", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -109,7 +109,7 @@ Deno.test( // But no disconnect packet should have been written. assertEquals(client.sentPackets.length, 0); - } + }, ); Deno.test("waiting for connack times out", async () => { @@ -212,7 +212,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -223,5 +223,5 @@ Deno.test( assertEquals(client.connectionState, "connected"); assertEquals(connectResolved, true); - } + }, ); diff --git a/client/publish_test.ts b/client/publish_test.ts index e76e778..0bb8d7f 100644 --- a/client/publish_test.ts +++ b/client/publish_test.ts @@ -31,7 +31,7 @@ Deno.test("publish qos 0 while connected", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -85,7 +85,7 @@ Deno.test("publish qos 0 while connecting", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -119,7 +119,7 @@ Deno.test("publish qos 1 while connected", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -148,7 +148,7 @@ Deno.test("publish qos 1 while connected", async () => { assertEquals(publish1Resolved, false); client.testReceiveBytes( - encodePuback({ type: "puback", id: publish1Packet.id! }) + encodePuback({ type: "puback", id: publish1Packet.id! }), ); await client.sleep(1); @@ -176,7 +176,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -223,7 +223,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -242,7 +242,7 @@ Deno.test( assertEquals(publish1Packet2.dup, true); client.testReceiveBytes( - encodePuback({ type: "puback", id: publish1Packet1.id! }) + encodePuback({ type: "puback", id: publish1Packet1.id! }), ); await client.sleep(1); @@ -266,7 +266,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -276,7 +276,7 @@ Deno.test( // ...but no new packets get sent. assertEquals(client.sentPackets.length, 5); - } + }, ); Deno.test( @@ -299,7 +299,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -346,7 +346,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -365,7 +365,7 @@ Deno.test( assertEquals(publish1Packet2.dup, true); client.testReceiveBytes( - encodePubrec({ type: "pubrec", id: publish1Packet1.id! }) + encodePubrec({ type: "pubrec", id: publish1Packet1.id! }), ); await client.sleep(1); @@ -394,7 +394,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -408,7 +408,7 @@ Deno.test( assertEquals(publish1Resolved, false); client.testReceiveBytes( - encodePubcomp({ type: "pubcomp", id: pubrel1Packet1.id }) + encodePubcomp({ type: "pubcomp", id: pubrel1Packet1.id }), ); await client.sleep(1); @@ -432,7 +432,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -442,7 +442,7 @@ Deno.test( // ...but no new packets get sent. assertEquals(client.sentPackets.length, 8); - } + }, ); Deno.test( @@ -468,7 +468,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -484,8 +484,8 @@ Deno.test( qos: 2, id: 12, }, - new TextEncoder() - ) + new TextEncoder(), + ), ); await client.sleep(1); @@ -512,7 +512,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -528,8 +528,8 @@ Deno.test( qos: 2, id: 12, }, - new TextEncoder() - ) + new TextEncoder(), + ), ); assertEquals(emittedMessages.length, 1); @@ -544,7 +544,7 @@ Deno.test( encodePubrel({ type: "pubrel", id: 12, - }) + }), ); await client.sleep(1); @@ -564,12 +564,12 @@ Deno.test( qos: 2, id: 13, }, - new TextEncoder() - ) + new TextEncoder(), + ), ); await client.sleep(1); assertEquals(emittedMessages.length, 2); - } + }, ); diff --git a/client/subscription_test.ts b/client/subscription_test.ts index e5936a2..3ca7197 100644 --- a/client/subscription_test.ts +++ b/client/subscription_test.ts @@ -25,7 +25,7 @@ Deno.test("subscribe and unsubscribe called while connected", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -61,7 +61,7 @@ Deno.test("subscribe and unsubscribe called while connected", async () => { type: "suback", id: subscribePacket.id, returnCodes: [0], - }) + }), ); assertEquals(subscribe1Resolved, false); @@ -116,7 +116,7 @@ Deno.test("subscribe and unsubscribe called while connected", async () => { ]); client.testReceiveBytes( - encodeUnsuback({ type: "unsuback", id: unsubscribePacket.id }) + encodeUnsuback({ type: "unsuback", id: unsubscribePacket.id }), ); // There are now no subscriptions. @@ -172,7 +172,7 @@ Deno.test("subscribe called while connecting", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -246,7 +246,7 @@ Deno.test("reconnecting resubscribes", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -276,7 +276,7 @@ Deno.test("reconnecting resubscribes", async () => { type: "suback", id: subscribe1Packet.id, returnCodes: [0], - }) + }), ); assertEquals(client.subscriptions, [ @@ -309,7 +309,7 @@ Deno.test("reconnecting resubscribes", async () => { type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -344,7 +344,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -375,7 +375,7 @@ Deno.test( type: "suback", id: subscribe1Packet.id, returnCodes: [0], - }) + }), ); assertEquals(client.subscriptions, [ @@ -409,7 +409,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: true, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -424,7 +424,7 @@ Deno.test( // But it doesn't get sent. assertEquals(client.sentPackets.length, 3); - } + }, ); Deno.test( @@ -448,7 +448,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -479,7 +479,7 @@ Deno.test( type: "suback", id: subscribe1Packet.id, returnCodes: [0], - }) + }), ); assertEquals(client.subscriptions, [ @@ -510,7 +510,7 @@ Deno.test( ]); assertEquals(unsubscribe1Result[0], topic1Subscription); - } + }, ); Deno.test( @@ -534,7 +534,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -565,7 +565,7 @@ Deno.test( type: "suback", id: subscribe1Packet.id, returnCodes: [0], - }) + }), ); assertEquals(client.subscriptions, [ @@ -616,7 +616,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: true, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -641,7 +641,7 @@ Deno.test( }); client.testReceiveBytes( - encodeUnsuback({ type: "unsuback", id: unsubscribe1Packet.id }) + encodeUnsuback({ type: "unsuback", id: unsubscribe1Packet.id }), ); assertEquals(unsubscribe1Resolved, false); @@ -663,7 +663,7 @@ Deno.test( }); assertEquals(unsubscribe1Result[0], topic1Subscription); - } + }, ); Deno.test( @@ -687,7 +687,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -720,7 +720,7 @@ Deno.test( type: "suback", id: subscribe1Packet.id, returnCodes: [0], - }) + }), ); assertEquals(client.subscriptions, [ @@ -771,7 +771,7 @@ Deno.test( type: "connack", returnCode: 0, sessionPresent: false, - }) + }), ); assertEquals(client.connectionState, "connected"); @@ -808,5 +808,5 @@ Deno.test( ]); assertEquals(unsubscribe1Result[0], topic1Subscription); - } + }, ); diff --git a/client/test_client.ts b/client/test_client.ts index d308bce..cc44f88 100644 --- a/client/test_client.ts +++ b/client/test_client.ts @@ -72,7 +72,7 @@ export class TestClient extends BaseClient { bytes: Uint8Array, options: { trickle?: boolean; - } = {} + } = {}, ) { if (options.trickle) { for (let i = 0; i < bytes.length; i++) { @@ -99,7 +99,7 @@ export class TestClient extends BaseClient { protected startTimer( name: string, cb: (...args: unknown[]) => void, - _delay: number + _delay: number, ) { this.timerCallbacks[name] = cb; } diff --git a/docker-compose.yaml b/docker-compose.yaml new file mode 100644 index 0000000..7aea204 --- /dev/null +++ b/docker-compose.yaml @@ -0,0 +1,8 @@ +services: + mqtt: + image: eclipse-mosquitto + volumes: + - ./mosquitto-docker.conf:/mosquitto/config/mosquitto.conf + ports: + - 1883:1883 + - 9001:9001 diff --git a/examples/deno/lwt.ts b/examples/deno/lwt.ts new file mode 100644 index 0000000..25f73d2 --- /dev/null +++ b/examples/deno/lwt.ts @@ -0,0 +1,39 @@ +#!/usr/bin/env deno run --allow-net + +import { Client } from '../../deno/mod.ts'; + +async function main() { + const client1 = new Client({ + url: 'mqtt://localhost:1883', + }); + + await client1.connect(); + console.log("Client 1 Connected") + + await client1.subscribe('lwt') + + client1.on('message', (topic: string, payload: Uint8Array) => { + console.log('Recieved:', topic, new TextDecoder().decode(payload)) + + client1.disconnect().then(() => void Deno.exit(0)) + }) + + + const client2 = new Client({ + url: 'mqtt://localhost:1883', + will: { + qos: 1, + retain: true, + topic: 'lwt', + payload: 'Client2 has crashed', + } + }) + + await client2.connect() + console.log("Client 2 Connected") + + //@ts-expect-error Accessing Private variable to 'crash' client 2 + client2.conn?.close() +} + +main(); diff --git a/examples/deno/pub.ts b/examples/deno/pub.ts index 523ccc9..1babdbb 100755 --- a/examples/deno/pub.ts +++ b/examples/deno/pub.ts @@ -1,6 +1,6 @@ #!/usr/bin/env deno run --allow-net -import { Client } from '../../mod.ts'; +import { Client } from '../../deno/mod.ts'; async function main() { const client = new Client({ diff --git a/examples/deno/sub.ts b/examples/deno/sub.ts index e5837a7..cd619e5 100755 --- a/examples/deno/sub.ts +++ b/examples/deno/sub.ts @@ -1,6 +1,6 @@ #!/usr/bin/env deno run --allow-net -import { Client } from '../../mod.ts'; +import { Client } from '../../deno/mod.ts'; async function main() { const client = new Client({ diff --git a/packets/binary.ts b/packets/binary.ts new file mode 100644 index 0000000..472574b --- /dev/null +++ b/packets/binary.ts @@ -0,0 +1,19 @@ +// Deno and browsers have global TextEncoder and TextDecoder classes, but +// Node.js does not so we have to use an abstraction for working with UTF8. + +export function encodeBinaryValue(bytes: Uint8Array) { + return [bytes.length >> 8, bytes.length & 0xff, ...bytes]; +} + +export function decodeBinaryValue( + buffer: Uint8Array, + startIndex: number, +) { + const length = (buffer[startIndex] << 8) + buffer[startIndex + 1]; + const bytes = buffer.slice(startIndex + 2, startIndex + 2 + length); + + return { + length: length + 2, + value: bytes, + }; +} diff --git a/packets/connack.ts b/packets/connack.ts index d62843e..7811c6e 100644 --- a/packets/connack.ts +++ b/packets/connack.ts @@ -19,7 +19,7 @@ export function encode(packet: ConnackPacket) { export function decode( buffer: Uint8Array, _remainingStart: number, - remainingLength: number + remainingLength: number, ): ConnackPacket { if (remainingLength !== 2) { throw new Error("connack packets must have a length of 2"); diff --git a/packets/connack_test.ts b/packets/connack_test.ts index b708860..d52ee97 100644 --- a/packets/connack_test.ts +++ b/packets/connack_test.ts @@ -19,7 +19,7 @@ Deno.test("encodeConnackPacket", function encodeConnackPacket() { // variableHeader 0, // connack flags 0, // return code - ] + ], ); }); @@ -35,13 +35,13 @@ Deno.test("decodeConnackPacket", function decodeConnackPacket() { 0, // return code ]), 2, - 2 + 2, ), { type: "connack", sessionPresent: false, returnCode: 0, - } + }, ); }); @@ -61,9 +61,9 @@ Deno.test( // variableHeader 1, // connack flags (sessionPresent) 0, // return code - ] + ], ); - } + }, ); Deno.test( @@ -80,15 +80,15 @@ Deno.test( 0, // return code ]), 2, - 2 + 2, ), { type: "connack", sessionPresent: true, returnCode: 0, - } + }, ); - } + }, ); Deno.test( @@ -105,15 +105,15 @@ Deno.test( 4, // return code (bad username or password) ]), 2, - 2 + 2, ), { type: "connack", sessionPresent: false, returnCode: 4, - } + }, ); - } + }, ); Deno.test("decodeShortConnackPackets", function decodeShortConnackPackets() { diff --git a/packets/connect.ts b/packets/connect.ts index c11c655..58f6026 100644 --- a/packets/connect.ts +++ b/packets/connect.ts @@ -1,4 +1,6 @@ +import { decodeBinaryValue, encodeBinaryValue } from "./binary.ts"; import { encodeLength } from "./length.ts"; +import { PublishPayload } from "./publish.ts"; import { decodeUTF8String, encodeUTF8String, @@ -14,8 +16,10 @@ export interface ConnectPacket { username?: string; password?: string; will?: { - retain?: boolean; - qos?: 0 | 1 | 2; + retain: boolean; + qos: 0 | 1 | 2; + topic: string; + payload: PublishPayload; }; clean?: boolean; keepAlive?: number; @@ -29,24 +33,21 @@ export function encode(packet: ConnectPacket, utf8Encoder: UTF8Encoder) { const protocolLevel = 4; const usernameFlag = !!packet.username; - const passwordFlag = !!packet.password; - const willRetain = !!(packet.will && packet.will.retain); - const willQoS = (packet.will && packet.will.qos) || 0; + const passwordFlag = usernameFlag && !!packet.password; + const willRetain = packet?.will?.retain ?? false; + const willQoS = packet?.will?.qos ?? 0; const willFlag = !!packet.will; const cleanSession = packet.clean || typeof packet.clean === "undefined"; - const connectFlags = - (usernameFlag ? 128 : 0) + + const connectFlags = (usernameFlag ? 128 : 0) + (passwordFlag ? 64 : 0) + (willRetain ? 32 : 0) + - (willQoS & 2 ? 16 : 0) + - (willQoS & 1 ? 8 : 0) + + (willQoS << 3) + (willFlag ? 4 : 0) + (cleanSession ? 2 : 0); - const keepAlive = - packet.keepAlive && typeof packet.keepAlive !== "undefined" - ? packet.keepAlive - : 0; + const keepAlive = packet.keepAlive && typeof packet.keepAlive !== "undefined" + ? packet.keepAlive + : 0; const variableHeader = [ ...protocolName, @@ -56,14 +57,26 @@ export function encode(packet: ConnectPacket, utf8Encoder: UTF8Encoder) { keepAlive & 0xff, ]; - const payload = [...encodeUTF8String(packet.clientId, utf8Encoder)]; + const encodeStr = (str: string) => encodeUTF8String(str, utf8Encoder); - if (packet.username) { - payload.push(...encodeUTF8String(packet.username, utf8Encoder)); + const payload = [...encodeStr(packet.clientId)]; + + if (typeof packet.will !== "undefined") { + payload.push(...encodeStr(packet.will.topic)); + + if (typeof packet.will.payload === "string") { + payload.push(...encodeStr(packet.will.payload)); + } else { + payload.push(...encodeBinaryValue(packet.will.payload)); + } } - if (packet.password) { - payload.push(...encodeUTF8String(packet.password, utf8Encoder)); + if (usernameFlag) { + payload.push(...encodeStr(packet.username!)); + } + + if (passwordFlag) { + payload.push(...encodeStr(packet.password!)); } const fixedHeader = [ @@ -78,16 +91,18 @@ export function decode( buffer: Uint8Array, remainingStart: number, _remainingLength: number, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ): ConnectPacket { - const protocolNameStart = remainingStart; - const protocolName = decodeUTF8String(buffer, protocolNameStart, utf8Decoder); + let readOffset = remainingStart; + + const protocolName = decodeUTF8String(buffer, readOffset, utf8Decoder); + readOffset += protocolName.length; - const protocolLevelIndex = protocolNameStart + protocolName.length; - const protocolLevel = buffer[protocolLevelIndex]; + const protocolLevel = buffer[readOffset]; + readOffset += 1; - const connectFlagsIndex = protocolLevelIndex + 1; - const connectFlags = buffer[connectFlagsIndex]; + const connectFlags = buffer[readOffset]; + readOffset += 1; const usernameFlag = !!(connectFlags & 128); const passwordFlag = !!(connectFlags & 64); const willRetain = !!(connectFlags & 32); @@ -99,24 +114,37 @@ export function decode( throw new Error("invalid will qos"); } - const keepAliveIndex = connectFlagsIndex + 1; - const keepAlive = (buffer[keepAliveIndex] << 8) + buffer[keepAliveIndex + 1]; + const keepAlive = (buffer[readOffset] << 8) + buffer[readOffset + 1]; + readOffset += 2; - const clientIdStart = keepAliveIndex + 2; - const clientId = decodeUTF8String(buffer, clientIdStart, utf8Decoder); + const clientId = decodeUTF8String(buffer, readOffset, utf8Decoder); + readOffset += clientId.length; - let username; - let password; + let willTopic: string | undefined = undefined; + let willPayload: Uint8Array | undefined = undefined; + let username: string | undefined = undefined; + let password: string | undefined = undefined; - const usernameStart = clientIdStart + clientId.length; + if (willFlag) { + const topic = decodeUTF8String(buffer, readOffset, utf8Decoder); + willTopic = topic.value; + readOffset += topic.length; + + const payload = decodeBinaryValue(buffer, readOffset); + willPayload = payload.value; + readOffset += topic.length; + } if (usernameFlag) { - username = decodeUTF8String(buffer, usernameStart, utf8Decoder); + const res = decodeUTF8String(buffer, readOffset, utf8Decoder); + readOffset += res.length; + username = res.value; } if (passwordFlag) { - const passwordStart = usernameStart + (username ? username.length : 0); - password = decodeUTF8String(buffer, passwordStart, utf8Decoder); + const res = decodeUTF8String(buffer, readOffset, utf8Decoder); + readOffset += res.length; + password = res.value; } return { @@ -124,13 +152,15 @@ export function decode( protocolName: protocolName.value, protocolLevel, clientId: clientId.value, - username: username ? username.value : undefined, - password: password ? password.value : undefined, + username: username, + password: password, will: willFlag ? { - retain: willRetain, - qos: willQoS, - } + retain: willRetain, + qos: willQoS, + payload: willPayload ?? "", + topic: willTopic ?? "", + } : undefined, clean: cleanSession, keepAlive, diff --git a/packets/connect_test.ts b/packets/connect_test.ts index 1429d88..d680774 100644 --- a/packets/connect_test.ts +++ b/packets/connect_test.ts @@ -13,7 +13,7 @@ Deno.test( type: "connect", clientId: "id", }, - utf8Encoder + utf8Encoder, ), [ // fixedHeader @@ -36,9 +36,9 @@ Deno.test( 2, // length LSB 105, // 'i' 100, // 'd' - ] + ], ); - } + }, ); Deno.test( @@ -51,7 +51,7 @@ Deno.test( clientId: "id", clean: false, }, - utf8Encoder + utf8Encoder, ), [ // fixedHeader @@ -74,9 +74,9 @@ Deno.test( 2, // length LSB 105, // 'i' 100, // 'd' - ] + ], ); - } + }, ); Deno.test( @@ -89,7 +89,7 @@ Deno.test( clientId: "id", keepAlive: 300, }, - utf8Encoder + utf8Encoder, ), [ // fixedHeader @@ -112,9 +112,9 @@ Deno.test( 2, // length LSB 105, // 'i' 100, // 'd' - ] + ], ); - } + }, ); Deno.test( @@ -128,7 +128,7 @@ Deno.test( username: "user", password: "pass", }, - utf8Encoder + utf8Encoder, ), [ // fixedHeader @@ -165,9 +165,9 @@ Deno.test( 97, // 'a' 115, // 's' 115, // 's' - ] + ], ); - } + }, ); Deno.test( @@ -213,7 +213,7 @@ Deno.test( ]), 2, 26, - new TextDecoder() + new TextDecoder(), ), { type: "connect", @@ -225,7 +225,141 @@ Deno.test( will: undefined, clean: true, keepAlive: 0, - } + }, ); - } + }, +); + +Deno.test( + "encodeConnectPacketWithLWT", + function encodeConnectPacketWithCleanFalse() { + assertEquals( + encode( + { + type: "connect", + clientId: "id", + will: { + qos: 1, + retain: true, + topic: "topic1", + payload: "offline", + }, + }, + utf8Encoder, + ), + [ + // fixedHeader + 16, // packetType + flags + 31, // remainingLength + // variableHeader + 0, // protocolNameLength MSB + 4, // protocolNameLength LSB + 77, // 'M' + 81, // 'Q' + 84, // 'T' + 84, // 'T' + 4, // protocolLevel + 2 + 4 + 8 + 32, // connectFlags (cleanSession, Will Flag, QOS 1, retain) + 0, // keepAlive MSB + 0, // keepAlive LSB + // payload + // clientId + 0, // length MSB + 2, // length LSB + 105, // 'i' + 100, // 'd' + // LWT topic + 0, // length MSB + 6, // length LSB + 116, // 't' + 111, // 'o' + 112, // 'p' + 105, // 'i' + 99, // 'c' + 49, // '1' + // LWT Payload + 0, // length MSB + 7, // length LSB + 111, // 'o' + 102, // 'f' + 102, // 'f' + 108, // 'l' + 105, // 'i' + 110, // 'n' + 101, // 'e' + ], + ); + }, +); + +Deno.test( + "decodeConnectPacketWithLWT", + function encodeConnectPacketWithCleanFalse() { + assertEquals( + decode( + Uint8Array.from( + [ + // fixedHeader + 16, // packetType + flags + 31, // remainingLength + // variableHeader + 0, // protocolNameLength MSB + 4, // protocolNameLength LSB + 77, // 'M' + 81, // 'Q' + 84, // 'T' + 84, // 'T' + 4, // protocolLevel + 2 + 4 + 8 + 32, // connectFlags (cleanSession, Will Flag, QOS 1, retain) + 0, // keepAlive MSB + 0, // keepAlive LSB + // payload + // clientId + 0, // length MSB + 2, // length LSB + 105, // 'i' + 100, // 'd' + // LWT topic + 0, // length MSB + 6, // length LSB + 116, // 't' + 111, // 'o' + 112, // 'p' + 105, // 'i' + 99, // 'c' + 49, // '1' + // LWT Payload + 0, // length MSB + 7, // length LSB + 111, // 'o' + 102, // 'f' + 102, // 'f' + 108, // 'l' + 105, // 'i' + 110, // 'n' + 101, // 'e' + ], + ), + 2, + 26, + new TextDecoder(), + ), + { + type: "connect", + protocolName: "MQTT", + protocolLevel: 4, + clientId: "id", + username: undefined, + password: undefined, + will: { + qos: 1, + retain: true, + topic: "topic1", + payload: utf8Encoder.encode("offline"), + }, + clean: true, + keepAlive: 0, + }, + ); + }, ); diff --git a/packets/disconnect.ts b/packets/disconnect.ts index 33fd503..cf4d269 100644 --- a/packets/disconnect.ts +++ b/packets/disconnect.ts @@ -12,7 +12,7 @@ export function encode(_packet: DisconnectPacket) { export function decode( _buffer: Uint8Array, _remainingStart: number, - _remainingLength: number + _remainingLength: number, ): DisconnectPacket { return { type: "disconnect", diff --git a/packets/disconnect_test.ts b/packets/disconnect_test.ts index 5f12045..e8b59d4 100644 --- a/packets/disconnect_test.ts +++ b/packets/disconnect_test.ts @@ -11,7 +11,7 @@ Deno.test("encodeDisconnectPacket", function encodeDisconnectPacket() { // fixedHeader 224, // packetType + flags 0, // remainingLength - ] + ], ); }); @@ -24,10 +24,10 @@ Deno.test("decodeDisconnectPacket", function decodeDisconnectPacket() { 0, // remainingLength ]), 2, - 0 + 0, ), { type: "disconnect", - } + }, ); }); diff --git a/packets/mod.ts b/packets/mod.ts index 4b95590..e468c36 100644 --- a/packets/mod.ts +++ b/packets/mod.ts @@ -66,14 +66,14 @@ export type { export type PacketEncoder = ( packet: T, - utf8Encoder: UTF8Encoder + utf8Encoder: UTF8Encoder, ) => Uint8Array; export type PacketDecoder = ( packet: Uint8Array, remainingStart: number, remainingLength: number, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ) => T; const packetDecoders = [ @@ -96,7 +96,7 @@ const packetDecoders = [ export function decode( buffer: Uint8Array, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ): AnyPacketWithLength | null { if (buffer.length < 2) { return null; @@ -112,7 +112,7 @@ export function decode( const { length: remainingLength, bytesUsedToEncodeLength } = decodeLength( buffer, - 1 + 1, ); const packetLength = 1 + bytesUsedToEncodeLength + remainingLength; @@ -125,7 +125,7 @@ export function decode( buffer, 1 + bytesUsedToEncodeLength, remainingLength, - utf8Decoder + utf8Decoder, ); if (!packet) { diff --git a/packets/pingreq.ts b/packets/pingreq.ts index 02b9e5c..4af2219 100644 --- a/packets/pingreq.ts +++ b/packets/pingreq.ts @@ -12,7 +12,7 @@ export function encode(_packet: PingreqPacket) { export function decode( _buffer: Uint8Array, _remainingStart: number, - _remainingLength: number + _remainingLength: number, ): PingreqPacket { return { type: "pingreq", diff --git a/packets/pingreq_test.ts b/packets/pingreq_test.ts index 7fa386d..235f264 100644 --- a/packets/pingreq_test.ts +++ b/packets/pingreq_test.ts @@ -11,7 +11,7 @@ Deno.test("encodePingreqPacket", function encodePingreqPacket() { // fixedHeader 0xc0, // packetType + flags 0, // remainingLength - ] + ], ); }); @@ -24,11 +24,11 @@ Deno.test("decodePingreqPacket", function decodePingreqPacket() { 0, // remainingLength ]), 2, - 0 + 0, ), { type: "pingreq", - } + }, ); }); diff --git a/packets/pingres.ts b/packets/pingres.ts index 2cbf3bc..0862647 100644 --- a/packets/pingres.ts +++ b/packets/pingres.ts @@ -9,7 +9,7 @@ export function encode(_packet: PingresPacket) { export function decode( _buffer: Uint8Array, _remainingStart: number, - _remainingLength: number + _remainingLength: number, ): PingresPacket { return { type: "pingres", diff --git a/packets/pingres_test.ts b/packets/pingres_test.ts index bd56f77..996abea 100644 --- a/packets/pingres_test.ts +++ b/packets/pingres_test.ts @@ -11,7 +11,7 @@ Deno.test("encodePingresPacket", function encodePingresPacket() { // fixedHeader 0xd0, // packetType + flags 0, // remainingLength - ] + ], ); }); @@ -24,11 +24,11 @@ Deno.test("decodePingresPacket", function decodePingresPacket() { 0, // remainingLength ]), 2, - 0 + 0, ), { type: "pingres", - } + }, ); }); diff --git a/packets/puback.ts b/packets/puback.ts index 89b2616..411c788 100644 --- a/packets/puback.ts +++ b/packets/puback.ts @@ -18,7 +18,7 @@ export function encode(packet: PubackPacket) { export function decode( buffer: Uint8Array, _remainingStart: number, - remainingLength: number + remainingLength: number, ): PubackPacket { if (remainingLength !== 2) { throw new Error("puback packets must have a length of 2"); diff --git a/packets/puback_test.ts b/packets/puback_test.ts index 1d10a81..f2f9e30 100644 --- a/packets/puback_test.ts +++ b/packets/puback_test.ts @@ -18,7 +18,7 @@ Deno.test("encodePubackPacket", function encodePubackPacket() { // variableHeader 5, // id MSB 57, // id LSB - ] + ], ); }); @@ -34,12 +34,12 @@ Deno.test("decodePubackPacket", function decodePubackPacket() { 57, // id LSB ]), 2, - 2 + 2, ), { type: "puback", id: 1337, - } + }, ); }); diff --git a/packets/pubcomp.ts b/packets/pubcomp.ts index 63a8be7..cd13033 100644 --- a/packets/pubcomp.ts +++ b/packets/pubcomp.ts @@ -18,7 +18,7 @@ export function encode(packet: PubcompPacket) { export function decode( buffer: Uint8Array, _remainingStart: number, - remainingLength: number + remainingLength: number, ): PubcompPacket { if (remainingLength !== 2) { throw new Error("pubcomp packets must have a length of 2"); diff --git a/packets/pubcomp_test.ts b/packets/pubcomp_test.ts index 4034a88..61a5c81 100644 --- a/packets/pubcomp_test.ts +++ b/packets/pubcomp_test.ts @@ -18,7 +18,7 @@ Deno.test("encodePubcompPacket", function encodePubcompPacket() { // variableHeader 5, // id MSB 57, // id LSB - ] + ], ); }); @@ -34,12 +34,12 @@ Deno.test("decodePubcompPacket", function decodePubcompPacket() { 57, // id LSB ]), 2, - 2 + 2, ), { type: "pubcomp", id: 1337, - } + }, ); }); diff --git a/packets/publish.ts b/packets/publish.ts index e7b613b..122f815 100644 --- a/packets/publish.ts +++ b/packets/publish.ts @@ -23,8 +23,7 @@ export function encode(packet: PublishPacket, utf8Encoder: UTF8Encoder) { const qos = packet.qos || 0; - const flags = - (packet.dup ? 8 : 0) + + const flags = (packet.dup ? 8 : 0) + (qos & 2 ? 4 : 0) + (qos & 1 ? 2 : 0) + (packet.retain ? 1 : 0); @@ -57,7 +56,7 @@ export function decode( buffer: Uint8Array, remainingStart: number, remainingLength: number, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ): PublishPacket { const flags = buffer[0] & 0x0f; diff --git a/packets/publish_test.ts b/packets/publish_test.ts index dafbdba..c241ab0 100644 --- a/packets/publish_test.ts +++ b/packets/publish_test.ts @@ -16,7 +16,7 @@ Deno.test("encodePublishPacket", function encodePublishPacket() { qos: 0, id: 0, }, - new TextEncoder() + new TextEncoder(), ), [ // fixedHeader @@ -36,7 +36,7 @@ Deno.test("encodePublishPacket", function encodePublishPacket() { 111, // 'o' 97, // 'a' 100, // 'd' - ] + ], ); }); @@ -64,7 +64,7 @@ Deno.test("decodePublishPacket", function decodePublishPacket() { ]), 2, 12, - utf8Decoder + utf8Decoder, ), { type: "publish", @@ -82,7 +82,7 @@ Deno.test("decodePublishPacket", function decodePublishPacket() { 97, // 'a' 100, // 'd' ]), - } + }, ); }); @@ -115,7 +115,7 @@ Deno.test( ]), 2, 12, - utf8Decoder + utf8Decoder, ), { type: "publish", @@ -133,7 +133,7 @@ Deno.test( 97, // 'a' 100, // 'd' ]), - } + }, ); - } + }, ); diff --git a/packets/pubrec.ts b/packets/pubrec.ts index 64c3ad6..db5fac6 100644 --- a/packets/pubrec.ts +++ b/packets/pubrec.ts @@ -18,7 +18,7 @@ export function encode(packet: PubrecPacket) { export function decode( buffer: Uint8Array, _remainingStart: number, - remainingLength: number + remainingLength: number, ): PubrecPacket { if (remainingLength !== 2) { throw new Error("pubrec packets must have a length of 2"); diff --git a/packets/pubrec_test.ts b/packets/pubrec_test.ts index 4e82971..4e00e0c 100644 --- a/packets/pubrec_test.ts +++ b/packets/pubrec_test.ts @@ -18,7 +18,7 @@ Deno.test("encodePubrecPacket", function encodePubrecPacket() { // variableHeader 5, // id MSB 57, // id LSB - ] + ], ); }); @@ -34,12 +34,12 @@ Deno.test("decodePubrecPacket", function decodePubrecPacket() { 57, // id LSB ]), 2, - 2 + 2, ), { type: "pubrec", id: 1337, - } + }, ); }); diff --git a/packets/pubrel.ts b/packets/pubrel.ts index adf1a13..ca453f0 100644 --- a/packets/pubrel.ts +++ b/packets/pubrel.ts @@ -18,7 +18,7 @@ export function encode(packet: PubrelPacket) { export function decode( buffer: Uint8Array, _remainingStart: number, - remainingLength: number + remainingLength: number, ): PubrelPacket { if (remainingLength !== 2) { throw new Error("pubrel packets must have a length of 2"); diff --git a/packets/pubrel_test.ts b/packets/pubrel_test.ts index 24ed0a7..91fdf53 100644 --- a/packets/pubrel_test.ts +++ b/packets/pubrel_test.ts @@ -18,7 +18,7 @@ Deno.test("encodePubrelPacket", function encodePubrelPacket() { // variableHeader 5, // id MSB 57, // id LSB - ] + ], ); }); @@ -34,12 +34,12 @@ Deno.test("decodePubrelPacket", function decodePubrelPacket() { 57, // id LSB ]), 2, - 2 + 2, ), { type: "pubrel", id: 1337, - } + }, ); }); diff --git a/packets/suback.ts b/packets/suback.ts index ff4425e..0de568c 100644 --- a/packets/suback.ts +++ b/packets/suback.ts @@ -20,7 +20,7 @@ export function encode(packet: SubackPacket) { export function decode( buffer: Uint8Array, remainingStart: number, - _remainingLength: number + _remainingLength: number, ): SubackPacket { const idStart = remainingStart; const id = (buffer[idStart] << 8) + buffer[idStart + 1]; diff --git a/packets/suback_test.ts b/packets/suback_test.ts index 4588252..1e9c1a3 100644 --- a/packets/suback_test.ts +++ b/packets/suback_test.ts @@ -19,7 +19,7 @@ Deno.test("encodeSubackPacket", function decodeSubackPacket() { // payload 0, 1, - ] + ], ); }); @@ -38,12 +38,12 @@ Deno.test("decodeSubackPacket", function decodeSubackPacket() { 1, ]), 2, - 4 + 4, ), { type: "suback", id: 1, returnCodes: [0, 1], - } + }, ); }); diff --git a/packets/subscribe.ts b/packets/subscribe.ts index d69002f..2eaab32 100644 --- a/packets/subscribe.ts +++ b/packets/subscribe.ts @@ -42,7 +42,7 @@ export function decode( buffer: Uint8Array, remainingStart: number, _remainingLength: number, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ): SubscribePacket { const idStart = remainingStart; const id = (buffer[idStart] << 8) + buffer[idStart + 1]; @@ -50,7 +50,7 @@ export function decode( const subscriptionsStart = idStart + 2; const subscriptions: Subscription[] = []; - for (let i = subscriptionsStart; i < buffer.length; ) { + for (let i = subscriptionsStart; i < buffer.length;) { const topicFilter = decodeUTF8String(buffer, i, utf8Decoder); i += topicFilter.length; diff --git a/packets/subscribe_test.ts b/packets/subscribe_test.ts index 590c279..8fbb5d2 100644 --- a/packets/subscribe_test.ts +++ b/packets/subscribe_test.ts @@ -13,7 +13,7 @@ Deno.test("encodeSubscribePacket", function encodeSubscribePacket() { { topicFilter: "c/d", qos: 1 }, ], }, - new TextEncoder() + new TextEncoder(), ), [ // fixedHeader @@ -35,7 +35,7 @@ Deno.test("encodeSubscribePacket", function encodeSubscribePacket() { 47, // '/' 100, // 'd' 1, // qos - ] + ], ); }); @@ -65,7 +65,7 @@ Deno.test("decodeSubscribePacket", function decodeSubscribePacket() { ]), 2, 14, - new TextDecoder() + new TextDecoder(), ), { type: "subscribe", @@ -74,6 +74,6 @@ Deno.test("decodeSubscribePacket", function decodeSubscribePacket() { { topicFilter: "a/b", qos: 0 }, { topicFilter: "c/d", qos: 1 }, ], - } + }, ); }); diff --git a/packets/unsuback.ts b/packets/unsuback.ts index ab30e96..2ed6445 100644 --- a/packets/unsuback.ts +++ b/packets/unsuback.ts @@ -18,7 +18,7 @@ export function encode(packet: UnsubackPacket) { export function decode( buffer: Uint8Array, _remainingStart: number, - _remainingLength: number + _remainingLength: number, ): UnsubackPacket { const id = (buffer[2] << 8) + buffer[3]; diff --git a/packets/unsuback_test.ts b/packets/unsuback_test.ts index 62498f8..8794939 100644 --- a/packets/unsuback_test.ts +++ b/packets/unsuback_test.ts @@ -15,7 +15,7 @@ Deno.test("encodeUnsubackPacket", function encodeUnsubackPacket() { // variableHeader 0, // id MSB 1, // id LSB - ] + ], ); }); @@ -31,11 +31,11 @@ Deno.test("decodeUnsubackPacket", function decodeUnsubackPacket() { 1, // id LSB ]), 2, - 2 + 2, ), { type: "unsuback", id: 1, - } + }, ); }); diff --git a/packets/unsubscribe.ts b/packets/unsubscribe.ts index 4a1f53f..fd41de4 100644 --- a/packets/unsubscribe.ts +++ b/packets/unsubscribe.ts @@ -36,7 +36,7 @@ export function decode( buffer: Uint8Array, remainingStart: number, _remainingLength: number, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ): UnsubscribePacket { const idStart = remainingStart; const id = (buffer[idStart] << 8) + buffer[idStart + 1]; @@ -44,7 +44,7 @@ export function decode( const topicFiltersStart = idStart + 2; const topicFilters: string[] = []; - for (let i = topicFiltersStart; i < buffer.length; ) { + for (let i = topicFiltersStart; i < buffer.length;) { const topicFilter = decodeUTF8String(buffer, i, utf8Decoder); i += topicFilter.length; diff --git a/packets/unsubscribe_test.ts b/packets/unsubscribe_test.ts index d9b5368..21e8e88 100644 --- a/packets/unsubscribe_test.ts +++ b/packets/unsubscribe_test.ts @@ -10,7 +10,7 @@ Deno.test("encodeUnsubscribePacket", function encodeUnsubscribePacket() { id: 1, topicFilters: ["a/b", "c/d"], }, - new TextEncoder() + new TextEncoder(), ), [ // fixedHeader @@ -30,7 +30,7 @@ Deno.test("encodeUnsubscribePacket", function encodeUnsubscribePacket() { 99, // 'c' 47, // '/' 100, // 'd' - ] + ], ); }); @@ -58,12 +58,12 @@ Deno.test("decodeUnsubscribePacket", function decodeUnsubscribePacket() { ]), 2, 12, - new TextDecoder() + new TextDecoder(), ), { type: "unsubscribe", id: 1, topicFilters: ["a/b", "c/d"], - } + }, ); }); diff --git a/packets/utf8.ts b/packets/utf8.ts index 1452092..645b03e 100644 --- a/packets/utf8.ts +++ b/packets/utf8.ts @@ -1,5 +1,6 @@ // Deno and browsers have global TextEncoder and TextDecoder classes, but // Node.js does not so we have to use an abstraction for working with UTF8. +import { decodeBinaryValue, encodeBinaryValue } from "./binary.ts"; export interface UTF8Encoder { encode: (str?: string | undefined) => Uint8Array; @@ -12,20 +13,19 @@ export interface UTF8Decoder { export function encodeUTF8String(str: string, encoder: UTF8Encoder) { const bytes = encoder.encode(str); - return [bytes.length >> 8, bytes.length & 0xff, ...bytes]; + return encodeBinaryValue(bytes); } export function decodeUTF8String( buffer: Uint8Array, startIndex: number, - utf8Decoder: UTF8Decoder + utf8Decoder: UTF8Decoder, ) { - const length = (buffer[startIndex] << 8) + buffer[startIndex + 1]; - const bytes = buffer.slice(startIndex + 2, startIndex + 2 + length); - const value = utf8Decoder.decode(bytes); + const { length, value } = decodeBinaryValue(buffer, startIndex); + const sValue = utf8Decoder.decode(value); return { - length: length + 2, - value, + length, + value: sValue, }; }