mqtts - npm Package Compare versions

Comparing version 0.1.17 to 1.0.0-alpha.0

dist/errors/end-of-stream.error.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.EndOfStreamError = void 0;
 const ts_custom_error_1 = require("ts-custom-error");
@@ -4,0 +5,0 @@ class EndOfStreamError extends ts_custom_error_1.CustomError {

dist/errors/index.d.ts

 export * from './end-of-stream.error';
 export * from './malformed-packet.error';
-export * from './invalid-direction.error';
 export * from './unexpected-packet.error';
+export * from './connect.error';
+export * from './subscribe.error';

dist/errors/index.js

 "use strict";
-function __export(m) {
-    for (var p in m) if (!exports.hasOwnProperty(p)) exports[p] = m[p];
-}
+var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    Object.defineProperty(o, k2, { enumerable: true, get: function() { return m[k]; } });
+}) : (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    o[k2] = m[k];
+}));
+var __exportStar = (this && this.__exportStar) || function(m, exports) {
+    for (var p in m) if (p !== "default" && !exports.hasOwnProperty(p)) __createBinding(exports, m, p);
+};
 Object.defineProperty(exports, "__esModule", { value: true });
-__export(require("./end-of-stream.error"));
-__export(require("./malformed-packet.error"));
-__export(require("./invalid-direction.error"));
-__export(require("./unexpected-packet.error"));
+__exportStar(require("./end-of-stream.error"), exports);
+__exportStar(require("./malformed-packet.error"), exports);
+__exportStar(require("./unexpected-packet.error"), exports);
+__exportStar(require("./connect.error"), exports);
+__exportStar(require("./subscribe.error"), exports);
 //# sourceMappingURL=index.js.map

dist/errors/malformed-packet.error.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.MalformedPacketError = void 0;
 const ts_custom_error_1 = require("ts-custom-error");
@@ -4,0 +5,0 @@ class MalformedPacketError extends ts_custom_error_1.CustomError {

dist/errors/unexpected-packet.error.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.UnexpectedPacketError = void 0;
 const ts_custom_error_1 = require("ts-custom-error");
@@ -4,0 +5,0 @@ class UnexpectedPacketError extends ts_custom_error_1.CustomError {

dist/flow/incoming.flows.d.ts

 import { MqttMessage } from '../mqtt.message';
 import { PacketFlowFunc } from './packet-flow';
-export declare function incomingPingFlow(): PacketFlowFunc<void>;
-export declare function incomingPublishFlow(message: MqttMessage, identifier?: number): PacketFlowFunc<MqttMessage>;
+import { DefaultPacketReadResultMap } from '../packets/packet-reader';
+import { DefaultPacketWriteOptions } from '../packets/packet-writer';
+export declare function incomingPublishFlow(message: MqttMessage, identifier: number): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, MqttMessage>;

dist/flow/incoming.flows.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const packets_1 = require("../packets");
+exports.incomingPublishFlow = void 0;
 const mqtt_utilities_1 = require("../mqtt.utilities");
-function incomingPingFlow() {
+const packet_writer_1 = require("../packets/packet-writer");
+const mqtt_constants_1 = require("../mqtt.constants");
+// PINGREQ is from client to server
+//
+// export function incomingPingFlow(): PacketFlowFunc<DefaultPacketReadResultMap, void> {
+//     return success => ({
+//         start: () => {
+//             success();
+//             return new PingResponsePacket();
+//         },
+//     });
+// }
+function incomingPublishFlow(message, identifier) {
     return success => ({
         start: () => {
-            success();
-            return new packets_1.PingResponsePacket();
-        },
-    });
-}
-exports.incomingPingFlow = incomingPingFlow;
-function incomingPublishFlow(message, identifier = -1) {
-    return success => ({
-        start: () => {
             let packet = undefined;
             let emit = true;
             if (message.qosLevel === 1) {
-                packet = new packets_1.PublishAckPacket();
+                packet = packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.PubAck, { identifier });
             }
             else if (message.qosLevel === 2) {
-                packet = new packets_1.PublishReceivedPacket();
+                packet = packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.PubRec, { identifier });
                 emit = false;
             }
-            if (packet)
-                packet.identifier = identifier;
             if (emit)
@@ -35,5 +36,3 @@ success(message);
             success(message);
-            const response = new packets_1.PublishCompletePacket();
-            response.identifier = identifier;
-            return response;
+            return packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.PubComp, { identifier });
         },
@@ -40,0 +39,0 @@ });

dist/flow/index.js

 "use strict";
-function __export(m) {
-    for (var p in m) if (!exports.hasOwnProperty(p)) exports[p] = m[p];
-}
+var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    Object.defineProperty(o, k2, { enumerable: true, get: function() { return m[k]; } });
+}) : (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    o[k2] = m[k];
+}));
+var __exportStar = (this && this.__exportStar) || function(m, exports) {
+    for (var p in m) if (p !== "default" && !exports.hasOwnProperty(p)) __createBinding(exports, m, p);
+};
 Object.defineProperty(exports, "__esModule", { value: true });
-__export(require("./incoming.flows"));
-__export(require("./outgoing.flows"));
+__exportStar(require("./packet-flow"), exports);
+__exportStar(require("./incoming.flows"), exports);
+__exportStar(require("./outgoing.flows"), exports);
 //# sourceMappingURL=index.js.map

dist/flow/outgoing.flows.d.ts

@@ -1,10 +0,12 @@
-import { ConnectRequestOptions } from '../packets';
+import { ConnectRequestOptions, ConnectResponsePacket, SubscribeReturnCode } from '../packets';
 import { MqttMessageOutgoing } from '../mqtt.message';
 import { MqttSubscription } from '../mqtt.types';
 import { PacketFlowFunc } from './packet-flow';
-export declare function outgoingConnectFlow(options: ConnectRequestOptions): PacketFlowFunc<ConnectRequestOptions>;
-export declare function outgoingDisconnectFlow(): PacketFlowFunc<void>;
-export declare function outgoingPingFlow(): PacketFlowFunc<void>;
-export declare function outgoingPublishFlow(message: MqttMessageOutgoing, _identifier?: number): PacketFlowFunc<MqttMessageOutgoing>;
-export declare function outgoingSubscribeFlow(subscription: MqttSubscription, identifier?: number): PacketFlowFunc<MqttSubscription>;
-export declare function outgoingUnsubscribeFlow(subscription: MqttSubscription, identifier?: number): PacketFlowFunc<void>;
+import { DefaultPacketReadResultMap } from '../packets/packet-reader';
+import { DefaultPacketWriteOptions } from '../packets/packet-writer';
+export declare function outgoingConnectFlow(options: ConnectRequestOptions): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, ConnectResponsePacket>;
+export declare function outgoingDisconnectFlow(): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, void>;
+export declare function outgoingPingFlow(): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, void>;
+export declare function outgoingPublishFlow(message: MqttMessageOutgoing, _identifier?: number): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, MqttMessageOutgoing>;
+export declare function outgoingSubscribeFlow(subscription: MqttSubscription, identifier?: number): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, SubscribeReturnCode>;
+export declare function outgoingUnsubscribeFlow(subscription: MqttSubscription, identifier?: number): PacketFlowFunc<DefaultPacketReadResultMap, DefaultPacketWriteOptions, void>;

dist/flow/outgoing.flows.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const packets_1 = require("../packets");
-const lodash_1 = require("lodash");
+exports.outgoingUnsubscribeFlow = exports.outgoingSubscribeFlow = exports.outgoingPublishFlow = exports.outgoingPingFlow = exports.outgoingDisconnectFlow = exports.outgoingConnectFlow = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
 const mqtt_utilities_1 = require("../mqtt.utilities");
+const packet_writer_1 = require("../packets/packet-writer");
+const mqtt_constants_1 = require("../mqtt.constants");
+const errors_1 = require("../errors");
 function outgoingConnectFlow(options) {
-    options = lodash_1.defaults(options, {
-        protocol: 3,
-        clientId: 'mqtt_' + lodash_1.random(1, 100000),
-        cleanSession: true,
+    const finalOptions = {
+        protocolLevel: 4,
+        clientId: 'mqtt_' + Math.round(Math.random() * 10e5),
+        clean: true,
         keepAlive: 60,
-    });
+        protocolName: 'MQTT',
+        ...options,
+    };
     return (success, error) => ({
-        start: () => new packets_1.ConnectRequestPacket(options),
+        start: () => packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.Connect, finalOptions),
         accept: mqtt_utilities_1.isConnAck,
-        next: (res) => (res.isSuccess ? success(options) : error(res.errorName)),
+        next: (res) => (res.isSuccess ? success(res) : error(res.errorName)),
     });
@@ -25,3 +29,3 @@ }
             success();
-            return new packets_1.DisconnectRequestPacket();
+            return packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.Disconnect);
         },
@@ -33,3 +37,3 @@ });
     return success => ({
-        start: () => new packets_1.PingRequestPacket(),
+        start: () => packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.PingReq),
         accept: mqtt_utilities_1.isPingResp,
@@ -41,14 +45,16 @@ next: () => success(),
 function outgoingPublishFlow(message, _identifier) {
-    const id = (_identifier !== null && _identifier !== void 0 ? _identifier : mqtt_packet_1.MqttPacket.generateIdentifier());
+    const id = _identifier !== null && _identifier !== void 0 ? _identifier : mqtt_packet_1.generateIdentifier();
     let receivedPubRec = false;
     return success => ({
         start: () => {
-            const packet = new packets_1.PublishRequestPacket(message.topic, message.payload);
-            packet.qosLevel = message.qosLevel || 0;
-            packet.duplicate = message.duplicate || false;
-            packet.retained = message.retained || false;
+            const packet = packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.Publish, {
+                topic: message.topic,
+                payload: mqtt_utilities_1.toBuffer(message.payload),
+                qos: message.qosLevel || 0,
+                retain: message.retained || false,
+                duplicate: message.duplicate || false,
+                identifier: message.qosLevel ? id : undefined,
+            });
             if (!message.qosLevel)
                 success(message);
-            else
-                packet.identifier = id;
             return packet;
@@ -76,3 +82,3 @@ },
                 receivedPubRec = true;
-                return new packets_1.PublishReleasePacket(id);
+                return packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.PubRel, { identifier: id });
             }
@@ -84,16 +90,16 @@ },
 function outgoingSubscribeFlow(subscription, identifier) {
-    const id = (identifier !== null && identifier !== void 0 ? identifier : mqtt_packet_1.MqttPacket.generateIdentifier());
+    const id = identifier !== null && identifier !== void 0 ? identifier : mqtt_packet_1.generateIdentifier();
     return (success, error) => ({
-        start: () => {
-            const packet = new packets_1.SubscribeRequestPacket(subscription.topic, subscription.qosLevel || 0);
-            packet.identifier = id;
-            return packet;
-        },
+        start: () => packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.Subscribe, {
+            identifier: id,
+            subscriptions: [{ qos: subscription.qosLevel || 0,
+                    topic: subscription.topic, }],
+        }),
         accept: (packet) => mqtt_utilities_1.isSubAck(packet) && packet.identifier === id,
         next: (packet) => {
-            if (packet.returnCodes.every(value => !packet.isError(value))) {
-                success(subscription);
+            if (!packet.anyError) {
+                success(packet.returnCodes[0]);
             }
             else {
-                error(`Failed to subscribe to ${subscription.topic}`);
+                error(new errors_1.SubscribeError(`Failed to subscribe to ${subscription.topic} - Return Codes: ${packet.returnCodes.join(', ')}`));
             }
@@ -105,10 +111,9 @@ },
 function outgoingUnsubscribeFlow(subscription, identifier) {
-    const id = (identifier !== null && identifier !== void 0 ? identifier : mqtt_packet_1.MqttPacket.generateIdentifier());
+    const id = identifier !== null && identifier !== void 0 ? identifier : mqtt_packet_1.generateIdentifier();
     return success => ({
-        start: () => {
-            const packet = new packets_1.UnsubscribeRequestPacket(subscription.topic);
-            packet.identifier = id;
-            return packet;
-        },
-        accept: (packet) => mqtt_utilities_1.isUnsubAck(packet) && packet.identifier === id,
+        start: () => packet_writer_1.defaultWrite(mqtt_constants_1.PacketType.Unsubscribe, {
+            identifier: id,
+            topics: [subscription.topic]
+        }),
+        accept: packet => mqtt_utilities_1.isUnsubAck(packet) && packet.identifier === id,
         next: () => success(),
@@ -115,0 +120,0 @@ });

dist/flow/packet-flow.d.ts

@@ -1,8 +0,10 @@
-import { MqttPacket } from '../mqtt.packet';
 import { Resolvers } from '../mqtt.utilities';
-export declare type PacketFlowFunc<T> = (success: (value: T) => void, error: (error: Error | string) => void) => PacketFlowCallbacks;
-export interface PacketFlowCallbacks {
-    start(): MqttPacket | void | undefined | null;
-    accept?(packet: MqttPacket): boolean | undefined | null;
-    next?(last: MqttPacket): MqttPacket | void | undefined | null;
+import { PacketType } from '../mqtt.constants';
+import { DefaultPacketReadResultMap, PacketReadResultMap } from '../packets/packet-reader';
+import { DefaultPacketWriteOptions, PacketWriteOptionsMap, WriteData } from '../packets/packet-writer';
+export declare type PacketFlowFunc<ReadMap extends PacketReadResultMap, WriteMap extends PacketWriteOptionsMap, TResult, TPacket = unknown> = (success: (value: TResult) => void, error: (error: Error | string) => void) => PacketFlowCallbacks<ReadMap, WriteMap, TPacket>;
+export interface PacketFlowCallbacks<ReadMap extends PacketReadResultMap = DefaultPacketReadResultMap, WriteMap extends PacketWriteOptionsMap = DefaultPacketWriteOptions, TPacket = unknown> {
+    start(): WriteData<WriteMap, PacketType> | void | undefined | null;
+    accept?(packet: TPacket): boolean | undefined;
+    next?(last: TPacket): WriteData<WriteMap, PacketType> | void | undefined;
 }
@@ -13,2 +15,3 @@ export interface PacketFlowData<T> {
     finished: boolean;
+    flowId: unknown;
 }

dist/index.js

 "use strict";
-function __export(m) {
-    for (var p in m) if (!exports.hasOwnProperty(p)) exports[p] = m[p];
-}
+var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    Object.defineProperty(o, k2, { enumerable: true, get: function() { return m[k]; } });
+}) : (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    o[k2] = m[k];
+}));
+var __exportStar = (this && this.__exportStar) || function(m, exports) {
+    for (var p in m) if (p !== "default" && !exports.hasOwnProperty(p)) __createBinding(exports, m, p);
+};
 Object.defineProperty(exports, "__esModule", { value: true });
-__export(require("./errors"));
-__export(require("./flow"));
-__export(require("./packets"));
-__export(require("./mqtt.client"));
-__export(require("./mqtt.constants"));
-__export(require("./mqtt.packet"));
-__export(require("./mqtt.parser"));
-__export(require("./packet-stream"));
-__export(require("./mqtt.utilities"));
-__export(require("./transport"));
+__exportStar(require("./errors"), exports);
+__exportStar(require("./flow"), exports);
+__exportStar(require("./packets"), exports);
+__exportStar(require("./mqtt.client"), exports);
+__exportStar(require("./mqtt.constants"), exports);
+__exportStar(require("./mqtt.message"), exports);
+__exportStar(require("./mqtt.packet"), exports);
+__exportStar(require("./mqtt.parser"), exports);
+__exportStar(require("./mqtt.types"), exports);
+__exportStar(require("./packet-stream"), exports);
+__exportStar(require("./mqtt.utilities"), exports);
+__exportStar(require("./transport"), exports);
 //# sourceMappingURL=index.js.map

dist/mqtt.client.d.ts

 /// <reference types="node" />
-import { Observable, Subject } from 'rxjs';
-import { ExecuteDelayed, ExecuteNextTick, ExecutePeriodically, IncomingListenMessage, ListenOptions, ListenSubscribeOptions, MqttClientConstructorOptions, MqttSubscription, RegisterClientOptions, Resolvable, StopExecuting } from './mqtt.types';
+import { ExecuteDelayed, ExecuteNextTick, ExecutePeriodically, IncomingListenMessage, ListenOptions, ListenSubscribeOptions, MqttClientConstructorOptions, MqttSubscription, RegisterClientOptions, Resolvable, StopExecuting, TimerRef } from './mqtt.types';
 import { PacketFlowData, PacketFlowFunc } from './flow';
-import { MqttParser } from './mqtt.parser';
+import { MqttParseResult, MqttTransformer } from './mqtt.parser';
 import { Transport } from './transport';
-import { MqttPacket } from './mqtt.packet';
-import { ConnectResponsePacket } from './packets';
-import { MqttMessage, MqttMessageOutgoing } from './mqtt.message';
-export declare class MqttClient {
+import { ConnectRequestOptions, SubscribeReturnCode } from './packets';
+import { MqttMessageOutgoing } from './mqtt.message';
+import { Writable } from 'stream';
+import { DefaultPacketReadResultMap, PacketReadResultMap, DefaultPacketWriteOptions, PacketWriteOptionsMap, PacketWriter } from './packets';
+import { PacketType } from './mqtt.constants';
+import { MqttBaseClient } from './mqtt.base-client';
+import { HandlerFn, MqttListener, RemoveHandlerFn } from './mqtt.listener';
+export declare class MqttClient<ReadMap extends PacketReadResultMap = DefaultPacketReadResultMap, WriteMap extends PacketWriteOptionsMap = DefaultPacketWriteOptions> extends MqttBaseClient<ReadMap, WriteMap> {
     private mqttDebug;
-    private packetDebug;
+    private receiveDebug;
     private pingDebug;
@@ -18,76 +21,44 @@ protected executeNextTick: ExecuteNextTick;
     protected executeDelayed: ExecuteDelayed;
-    /**
-     * An error has been encountered, the client will no longer work correctly
-     * @type {Subject<Error>}
-     */
-    $error: Subject<Error>;
-    /**
-     * An error has been encountered, the client might still continue to work
-     * @type {Subject<Error>}
-     */
-    $warning: Subject<Error>;
-    /**
-     *
-     * @type {Subject<void>}
-     */
-    $open: Subject<void>;
-    /**
-     * The client successfully established a connection
-     * @type {Subject<void>}
-     */
-    $connect: Subject<ConnectResponsePacket>;
-    /**
-     * The client disconnected.
-     * @type {Subject<void>}
-     */
-    $disconnect: Subject<string | undefined>;
-    $message: Subject<MqttMessage>;
     get keepAlive(): number;
     set keepAlive(value: number);
     protected transport: Transport<unknown>;
-    protected parser: MqttParser;
-    protected connectTimer?: object;
-    protected keepAliveTimer?: object;
+    protected transformer: MqttTransformer<ReadMap>;
+    protected createTransformer: () => MqttTransformer<ReadMap>;
+    protected pipeline?: Writable;
+    protected writer: PacketWriter<WriteMap>;
+    protected connectTimer?: TimerRef;
+    protected keepAliveTimer?: TimerRef;
     protected autoReconnect: boolean;
-    protected state: MqttClientState;
     protected activeFlows: PacketFlowData<any>[];
-    constructor(options: MqttClientConstructorOptions);
+    protected messageListener: MqttListener;
+    constructor(options: MqttClientConstructorOptions<ReadMap, WriteMap>);
     connect(options?: Resolvable<RegisterClientOptions>): Promise<any>;
-    protected getConnectOptions(): Promise<RegisterClientOptions>;
-    protected registerClient(options: RegisterClientOptions, noNewPromise?: boolean): Promise<any>;
-    protected getConnectFlow(options: any): PacketFlowFunc<any>;
+    protected registerClient(options: RegisterClientOptions, noNewPromise?: boolean, lastFlow?: PacketFlowFunc<ReadMap, WriteMap, unknown>): Promise<any>;
+    protected getConnectFlow(options: ConnectRequestOptions): PacketFlowFunc<ReadMap, WriteMap, unknown>;
     publish(message: MqttMessageOutgoing): Promise<MqttMessageOutgoing>;
-    subscribe(subscription: MqttSubscription): Promise<MqttSubscription>;
+    subscribe(subscription: MqttSubscription): Promise<SubscribeReturnCode>;
     unsubscribe(subscription: MqttSubscription): Promise<void>;
-    disconnect(): Promise<void>;
-    listenSubscribe<T = IncomingListenMessage<any>>(topic: string): Promise<Observable<T>>;
-    listenSubscribe<T = IncomingListenMessage<any>>(options: ListenSubscribeOptions<T>): Promise<Observable<T>>;
-    listen<T>(topic: string): Observable<T>;
-    listen<T>(options: ListenOptions<T>): Observable<T>;
-    startFlow<T>(flow: PacketFlowFunc<T>): Promise<T>;
+    disconnect(force?: boolean): Promise<void>;
+    listenSubscribe<T = IncomingListenMessage>(topic: string, handlerFn: HandlerFn<T>): Promise<RemoveHandlerFn>;
+    listenSubscribe<T = IncomingListenMessage, Params extends Record<string, string> = Record<string, string>>(options: ListenSubscribeOptions<T, Params>, handlerFn: HandlerFn<T>): Promise<RemoveHandlerFn>;
+    listen<T>(topic: string, handlerFn: HandlerFn<T>): RemoveHandlerFn;
+    listen<T, Params extends Record<string, string>>(options: ListenOptions<T, Params>, handlerFn: HandlerFn<T>): RemoveHandlerFn;
+    startFlow<T>(flow: PacketFlowFunc<ReadMap, WriteMap, T>): Promise<T>;
     /**
-     *
+     *  Run the accept and next function of all active flows
      * @param {MqttPacket} packet
      * @returns {boolean} true if a flow has been found
      */
-    protected continueFlows(packet: MqttPacket): boolean;
-    protected checkFlows(): void;
+    protected continueFlows(packet: MqttParseResult<ReadMap, typeof PacketType[keyof typeof PacketType]>): boolean;
+    protected clearFinishedFlows(): void;
     protected updateKeepAlive(value: number): void;
-    protected sendPacket(packet: MqttPacket): void;
-    protected parseData(data: Buffer): Promise<void>;
-    protected handlePacket(packet: MqttPacket): Promise<void>;
-    protected logPacket(packet: MqttPacket, action: string): void;
+    protected sendData(data: Buffer): void;
+    protected handlePacket(packet: MqttParseResult<ReadMap, PacketType>): Promise<void>;
+    protected logReceivedPacket(packet: {
+        type: PacketType;
+        data: any;
+    }): void;
     protected reset(): void;
-    protected setConnecting(): void;
-    protected setConnected(): void;
-    protected setDisconnected(reason?: string): void;
+    protected setReady(): void;
+    protected setDisconnected(reason?: string): Promise<void>;
 }
-export interface MqttClientState {
-    connected: boolean;
-    connecting: boolean;
-    disconnected: boolean;
-    connectOptions?: RegisterClientOptions;
-    connectOptionsResolver?: Resolvable<RegisterClientOptions>;
-    startResolve?: () => void;
-    startReject?: (e: any) => void;
-}

dist/mqtt.client.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const rxjs_1 = require("rxjs");
+exports.MqttClient = void 0;
 const flow_1 = require("./flow");
 const mqtt_parser_1 = require("./mqtt.parser");
 const transport_1 = require("./transport");
-const lodash_1 = require("lodash");
-const packet_stream_1 = require("./packet-stream");
+const errors_1 = require("./errors");
+const stream_1 = require("stream");
+const packets_1 = require("./packets");
 const mqtt_constants_1 = require("./mqtt.constants");
-const operators_1 = require("rxjs/operators");
 const debug = require("debug");
+const mqtt_base_client_1 = require("./mqtt.base-client");
+const mqtt_listener_1 = require("./mqtt.listener");
 const mqtt_utilities_1 = require("./mqtt.utilities");
-const errors_1 = require("./errors");
-class MqttClient {
+class MqttClient extends mqtt_base_client_1.MqttBaseClient {
     constructor(options) {
-        var _a, _b, _c;
+        var _a, _b, _c, _d;
+        super();
         this.mqttDebug = debug('mqtt:client');
-        this.packetDebug = this.mqttDebug.extend('packet');
+        this.receiveDebug = this.mqttDebug.extend('packet');
         this.pingDebug = this.mqttDebug.extend('ping');
-        /**
-         * An error has been encountered, the client will no longer work correctly
-         * @type {Subject<Error>}
-         */
-        this.$error = new rxjs_1.Subject();
-        /**
-         * An error has been encountered, the client might still continue to work
-         * @type {Subject<Error>}
-         */
-        this.$warning = new rxjs_1.Subject();
-        /**
-         *
-         * @type {Subject<void>}
-         */
-        this.$open = new rxjs_1.Subject();
-        /**
-         * The client successfully established a connection
-         * @type {Subject<void>}
-         */
-        this.$connect = new rxjs_1.Subject();
-        /**
-         * The client disconnected.
-         * @type {Subject<void>}
-         */
-        this.$disconnect = new rxjs_1.Subject();
-        this.$message = new rxjs_1.Subject();
+        // wrapper functions
+        this.executeNextTick = process.nextTick;
+        this.executePeriodically = (ms, cb) => setInterval(cb, ms);
+        this.stopExecuting = clearInterval;
+        this.executeDelayed = (ms, cb) => setTimeout(cb, ms);
         this.activeFlows = [];
-        this.state = {
-            connected: false,
-            connecting: false,
-            disconnected: false,
-        };
+        this.messageListener = new mqtt_listener_1.MqttListener();
         this.autoReconnect = !!options.autoReconnect;
-        this.parser = (_a = options.parser, (_a !== null && _a !== void 0 ? _a : new mqtt_parser_1.MqttParser(e => this.$error.next(e), this.mqttDebug.extend('parser'))));
-        this.transport = (_b = options.transport, (_b !== null && _b !== void 0 ? _b : new transport_1.TlsTransport({
-            url: options.url,
-            enableTrace: (_c = options.enableTrace, (_c !== null && _c !== void 0 ? _c : false)),
-        })));
-        try {
-            this.executeNextTick = process.nextTick;
-            this.executePeriodically = (ms, cb) => setInterval(cb, ms);
-            this.executeDelayed = (ms, cb) => setTimeout(cb, ms);
-            this.stopExecuting = clearInterval;
-        }
-        catch (e) {
-            this.mqttDebug(`Could not register timers: ${e.stack}`);
-        }
+        this.transport = (_a = options.transport) !== null && _a !== void 0 ? _a : new transport_1.TlsTransport({
+            host: options.host,
+            port: options.port,
+            additionalOptions: {
+                enableTrace: options.enableTrace,
+            },
+        });
+        this.createTransformer = (_b = options.createTransformer) !== null && _b !== void 0 ? _b : (() => {
+            var _a;
+            return new mqtt_parser_1.MqttTransformer({
+                debug: this.mqttDebug.extend('transformer'),
+                mapping: (_a = options.readMap) !== null && _a !== void 0 ? _a : packets_1.DefaultPacketReadMap,
+            });
+        });
+        this.transformer = this.createTransformer();
+        this.transformer.options.debug = (_c = this.transformer.options.debug) !== null && _c !== void 0 ? _c : this.mqttDebug.extend('transformer');
+        const packetLogger = this.mqttDebug.extend('write');
+        this.writer = (_d = options.packetWriter) !== null && _d !== void 0 ? _d : new packets_1.PacketWriter({
+            logPacketWrite: mqtt_utilities_1.createDefaultPacketLogger(packetLogger),
+        }, options.writeMap);
     }
     get keepAlive() {
-        var _a, _b, _c;
-        return _c = (_b = (_a = this.state) === null || _a === void 0 ? void 0 : _a.connectOptions) === null || _b === void 0 ? void 0 : _b.keepAlive, (_c !== null && _c !== void 0 ? _c : 0);
+        var _a, _b;
+        return (_b = (_a = this.connectOptions) === null || _a === void 0 ? void 0 : _a.keepAlive) !== null && _b !== void 0 ? _b : 0;
     }
     set keepAlive(value) {
-        var _a;
-        if ((_a = this.state) === null || _a === void 0 ? void 0 : _a.connectOptions) {
-            this.state.connectOptions.keepAlive = value;
+        if (this.connectOptions) {
+            this.connectOptions.keepAlive = value;
             if (value) {
@@ -81,58 +63,59 @@ this.updateKeepAlive(value);
     }
-    connect(options) {
-        var _a;
-        if (this.state.connected || this.state.connecting) {
-            throw new Error('Invalid State: The client is already connecting/connected!');
-        }
+    async connect(options) {
+        this.expectCreated();
         this.mqttDebug('Connecting...');
-        this.state.connectOptionsResolver = (_a = this.state.connectOptionsResolver, (_a !== null && _a !== void 0 ? _a : options));
+        this.connectResolver = options;
         this.setConnecting();
-        return new Promise(resolve => {
-            this.transport.callbacks = {
-                disconnect: (data) => {
-                    var _a, _b;
-                    if (data) {
-                        this.mqttDebug(`Transport disconnected with ${data}\n${data.stack}`);
-                        this.$error.next(data);
-                    }
-                    this.setDisconnected(`error in transport ${(_a = data) === null || _a === void 0 ? void 0 : _a.name} ${(_b = data) === null || _b === void 0 ? void 0 : _b.stack}`);
-                },
-                connect: () => {
-                    this.$open.next();
-                    resolve();
-                },
-                error: (e) => this.$error.next(e),
-                data: (data) => this.parseData(data),
-            };
-            this.transport.connect();
-        }).then(async () => this.registerClient(await this.getConnectOptions()));
+        this.pipeline = stream_1.pipeline(this.transport.duplex, this.transformer, new stream_1.Writable({
+            write: (chunk, encoding, callback) => {
+                if (!chunk.type) {
+                    callback(new Error('Chunk is not a MqttPacket'));
+                    return;
+                }
+                this.handlePacket(chunk)
+                    .then(() => callback())
+                    .catch(callback);
+            },
+            objectMode: true,
+        }), err => {
+            if (err)
+                this.emitError(err);
+            if (!this.disconnected)
+                this.setDisconnected('Pipeline finished');
+        });
+        await this.transport.connect();
+        return this.registerClient(await this.resolveConnectOptions());
     }
-    async getConnectOptions() {
+    registerClient(options, noNewPromise = false, lastFlow) {
         var _a;
-        return (this.state.connectOptions = lodash_1.defaults(await mqtt_utilities_1.resolve(this.state.connectOptionsResolver || {}), (_a = this.state.connectOptions, (_a !== null && _a !== void 0 ? _a : {}))));
-    }
-    registerClient(options, noNewPromise = false) {
-        var _a;
         let promise;
         if (noNewPromise) {
-            promise = this.startFlow(this.getConnectFlow(options));
+            const flow = this.activeFlows.find(x => x.flowId === lastFlow);
+            if (!flow) {
+                promise = Promise.reject(new Error('Could not find flow'));
+            }
+            else {
+                const packet = flow.callbacks.start();
+                if (packet)
+                    this.sendData(this.writer.write(packet.type, packet.options));
+                promise = Promise.resolve();
+            }
         }
         else {
-            promise = new Promise((resolve, reject) => {
-                this.state.startResolve = resolve;
-                this.state.startReject = reject;
-            });
-            this.startFlow(this.getConnectFlow(options))
-                .then(() => { var _a, _b; return (_b = (_a = this.state).startResolve) === null || _b === void 0 ? void 0 : _b.call(_a); })
-                .catch(e => { var _a, _b; return (_b = (_a = this.state).startReject) === null || _b === void 0 ? void 0 : _b.call(_a, e); });
+            promise = this.createConnectPromise();
+            lastFlow = lastFlow !== null && lastFlow !== void 0 ? lastFlow : this.getConnectFlow(options);
+            this.startFlow(lastFlow)
+                .then(() => this.resolveConnectPromise())
+                .catch(e => this.rejectConnectPromise(e));
         }
         this.connectTimer =
-            options.connectDelay === null
+            typeof options.connectDelay === 'undefined'
                 ? undefined
-                : this.executeDelayed((_a = options.connectDelay, (_a !== null && _a !== void 0 ? _a : 2000)), () => this.registerClient(options, true)
-                    .then(() => { var _a, _b; return (_b = (_a = this.state).startResolve) === null || _b === void 0 ? void 0 : _b.call(_a); })
-                    .catch(e => { var _a, _b; return (_b = (_a = this.state).startReject) === null || _b === void 0 ? void 0 : _b.call(_a, e); }));
+                : this.executeDelayed((_a = options.connectDelay) !== null && _a !== void 0 ? _a : 2000, () => 
+                // This Promise will only reject if the flow wasn't found
+                this.registerClient(options, true, lastFlow).catch(e => this.rejectConnectPromise(e)));
         return promise;
     }
     getConnectFlow(options) {
+        // assume the defaults are used
         return flow_1.outgoingConnectFlow(options);
@@ -149,7 +132,12 @@ }
     }
-    disconnect() {
+    async disconnect(force = false) {
         this.autoReconnect = false;
-        return this.startFlow(flow_1.outgoingDisconnectFlow());
+        if (!force) {
+            return this.startFlow(flow_1.outgoingDisconnectFlow()).then(async () => await this.setDisconnected());
+        }
+        else {
+            await this.setDisconnected('Forced Disconnect');
+        }
     }
-    listenSubscribe(options) {
+    listenSubscribe(options, handlerFn) {
         const listener = typeof options === 'string' ? { topic: options } : options;
@@ -159,24 +147,14 @@ return this.subscribe({
             topic: listener.topic.replace(/\/:[A-Za-z-_0-9]+/g, '/+'),
-        }).then(() => this.listen(listener));
+        }).then(() => this.listen(listener, handlerFn));
     }
-    listen(options) {
+    listen(options, handlerFn) {
         const listener = typeof options === 'string' ? { topic: options } : options;
-        const paramRegex = /\/:[A-Za-z-_0-9]+/g;
-        let baseTopic = listener.topic;
-        if (listener.topic.match(paramRegex)) {
-            baseTopic = listener.topic.replace(paramRegex, '/+');
-        }
-        return this.$message.pipe(operators_1.filter(v => {
-            if (!mqtt_utilities_1.matchTopic(baseTopic, v.topic))
-                return false;
-            if (listener.validator === null)
-                return true;
-            if (!listener.validator) {
-                return !!v.payload;
-            }
-            return listener.validator(v);
-        }), operators_1.map((v) => {
-            v.params = mqtt_utilities_1.extractParams(listener.topic, v.topic);
-            return listener.transformer ? listener.transformer(v) : v;
-        }));
+        const [topicFilter, paramMatcher] = mqtt_utilities_1.toMqttTopicFilter(listener.topic);
+        return this.messageListener.addHandler({
+            topicFilter,
+            handle: handlerFn,
+            transformer: listener.transformer,
+            validator: listener.validator,
+            paramMatcher,
+        });
     }
@@ -195,6 +173,7 @@ startFlow(flow) {
                 }),
+                flowId: flow,
             };
             const first = data.callbacks.start();
             if (first)
-                this.sendPacket(first);
+                this.sendData(this.writer.write(first.type, first.options));
             if (!data.finished) {
@@ -206,3 +185,3 @@ this.activeFlows.push(data);
     /**
-     *
+     *  Run the accept and next function of all active flows
      * @param {MqttPacket} packet
@@ -215,6 +194,6 @@ * @returns {boolean} true if a flow has been found
         for (const flow of this.activeFlows) {
-            if ((_b = (_a = flow.callbacks).accept) === null || _b === void 0 ? void 0 : _b.call(_a, packet)) {
-                const next = (_d = (_c = flow.callbacks).next) === null || _d === void 0 ? void 0 : _d.call(_c, packet);
+            if ((_b = (_a = flow.callbacks).accept) === null || _b === void 0 ? void 0 : _b.call(_a, packet.data)) {
+                const next = (_d = (_c = flow.callbacks).next) === null || _d === void 0 ? void 0 : _d.call(_c, packet.data);
                 if (next) {
-                    this.sendPacket(next);
+                    this.sendData(this.writer.write(next.type, next.options));
                 }
@@ -224,7 +203,7 @@ result = true;
         }
-        this.checkFlows();
+        this.clearFinishedFlows();
         return result;
     }
-    checkFlows() {
-        this.activeFlows = lodash_1.pull(this.activeFlows, ...this.activeFlows.filter(f => f.finished));
+    clearFinishedFlows() {
+        this.activeFlows = this.activeFlows.filter(flow => !flow.finished);
     }
@@ -238,56 +217,54 @@ updateKeepAlive(value) {
         this.keepAliveTimer = this.executePeriodically(value * 1000, () => {
+            // assume the defaults are used
             this.startFlow(flow_1.outgoingPingFlow())
                 .then(() => this.pingDebug(`PingPong @ ${Date.now()}`))
-                .catch(() => this.pingDebug('PingPong failed.'));
+                .catch(e => {
+                this.emitWarning(e);
+                this.pingDebug(`PingPong failed. (${e.message})`);
+            });
         });
     }
-    sendPacket(packet) {
-        const stream = packet_stream_1.PacketStream.empty();
-        packet.write(stream);
-        this.logPacket(packet, 'Sent');
-        this.transport.send(stream.data);
+    sendData(data) {
+        this.transport.duplex.write(data);
     }
-    async parseData(data) {
-        try {
-            const results = await this.parser.parse(data);
-            if (results.length > 0) {
-                results.forEach(r => this.handlePacket(r));
-            }
-        }
-        catch (e) {
-            this.$warning.next(e);
-        }
-    }
     async handlePacket(packet) {
-        var _a, _b, _c;
-        this.logPacket(packet, 'Received');
+        var _a, _b;
+        this.logReceivedPacket(packet);
         let forceCheckFlows = false;
-        switch (packet.packetType) {
-            case mqtt_constants_1.PacketTypes.TYPE_CONNACK: {
+        switch (packet.type) {
+            case mqtt_constants_1.PacketType.ConnAck: {
                 const connack = packet;
-                if (connack.isSuccess) {
-                    this.setConnected();
-                    this.$connect.next(connack);
-                    if ((_b = (_a = this.state) === null || _a === void 0 ? void 0 : _a.connectOptions) === null || _b === void 0 ? void 0 : _b.keepAlive) {
-                        this.updateKeepAlive(this.state.connectOptions.keepAlive);
+                if (connack.data.isSuccess) {
+                    this.setReady();
+                    this.emitConnect(connack.data);
+                    if ((_a = this.connectOptions) === null || _a === void 0 ? void 0 : _a.keepAlive) {
+                        this.updateKeepAlive(this.connectOptions.keepAlive);
                     }
                 }
+                else {
+                    this.setFatal();
+                    this.emitError(new errors_1.ConnectError(connack.data.errorName));
+                    this.setDisconnected(connack.data.errorName).catch(e => this.emitWarning(e));
+                }
                 break;
             }
-            case mqtt_constants_1.PacketTypes.TYPE_PUBLISH: {
-                const pub = packet;
+            case mqtt_constants_1.PacketType.Publish: {
+                const pub = packet.data;
                 this.startFlow(flow_1.incomingPublishFlow({
                     topic: pub.topic,
                     payload: pub.payload,
-                    qosLevel: pub.qosLevel,
-                    retained: pub.retained,
+                    qosLevel: pub.qos,
+                    retained: pub.retain,
                     duplicate: pub.duplicate,
-                }, (_c = pub.identifier, (_c !== null && _c !== void 0 ? _c : undefined))))
-                    .then(m => this.$message.next(m))
-                    .catch(e => this.$warning.next(e));
+                }, (_b = pub.identifier) !== null && _b !== void 0 ? _b : undefined))
+                    .then(async (m) => {
+                    this.emitMessage(m);
+                    await this.messageListener.handleMessage(m);
+                })
+                    .catch(e => this.emitWarning(e));
                 break;
             }
-            case mqtt_constants_1.PacketTypes.TYPE_DISCONNECT: {
+            case mqtt_constants_1.PacketType.Disconnect: {
                 // ? this.disconnect();
-                this.setDisconnected('disconnect packet received');
+                this.setDisconnected('disconnect packet received').catch(e => this.emitWarning(e));
                 break;
@@ -299,50 +276,48 @@ }
         if (!this.continueFlows(packet) && forceCheckFlows) {
-            this.$warning.next(new errors_1.UnexpectedPacketError(packet.constructor.name));
+            this.emitWarning(new errors_1.UnexpectedPacketError(packet.constructor.name));
         }
     }
-    logPacket(packet, action) {
-        if (packet.packetType !== mqtt_constants_1.PacketTypes.TYPE_PINGREQ && packet.packetType !== mqtt_constants_1.PacketTypes.TYPE_PINGRESP)
-            this.packetDebug(`${action} ${packet.constructor.name}` +
-                (packet.identifier ? ` id: ${packet.identifier}` : '') +
-                // @ts-ignore - instanceof is too expensive
-                (packet.topic ? ` topic: ${packet.topic}` : ''));
+    logReceivedPacket(packet) {
+        if (packet.type !== mqtt_constants_1.PacketType.PingReq && packet.type !== mqtt_constants_1.PacketType.PingResp)
+            this.receiveDebug(`Received ${mqtt_utilities_1.stringifyObject(packet.data)}`);
     }
     reset() {
+        super.reset();
+        if (this.connecting)
+            this.rejectConnectPromise(new Error('Disconnected'));
         if (this.connectTimer)
-            this.stopExecuting(this.connectTimer);
+            clearTimeout(this.connectTimer);
         this.connectTimer = undefined;
         if (this.keepAliveTimer)
-            this.stopExecuting(this.keepAliveTimer);
+            clearInterval(this.keepAliveTimer);
         this.keepAliveTimer = undefined;
         this.activeFlows = [];
-        this.state.startResolve = undefined;
-        this.state.startReject = undefined;
-        this.parser.reset();
+        this.transformer.reset();
     }
-    setConnecting() {
-        this.state.connecting = true;
-        this.state.connected = false;
-        this.state.disconnected = false;
-    }
-    setConnected() {
-        this.mqttDebug('Connected!');
-        this.state.connecting = false;
-        this.state.connected = true;
-        this.state.disconnected = false;
+    setReady() {
+        super.setReady();
+        this.mqttDebug('Ready!');
         if (this.connectTimer)
             this.stopExecuting(this.connectTimer);
     }
-    setDisconnected(reason) {
-        const willReconnect = !this.state.disconnected && this.state.connected && !this.state.connecting && this.autoReconnect;
-        if (!this.state.disconnected) {
-            this.$disconnect.next(reason);
+    async setDisconnected(reason) {
+        var _a;
+        const willReconnect = !this.disconnected && this.ready && !this.connecting && this.autoReconnect;
+        if (this.connecting)
+            this.rejectConnectPromise(new Error('Disconnected'));
+        super.setDisconnected();
+        this.emitDisconnect(reason);
+        if (!this.transport.duplex.destroyed) {
+            await new Promise(resolve => this.transport.duplex.end(resolve));
+            if (!this.transport.duplex.writableEnded) {
+                this.transport.duplex.destroy(new Error('force destroy'));
+            }
         }
-        this.transport.disconnect();
-        this.state.connecting = false;
-        this.state.connected = false;
-        this.state.disconnected = true;
         this.stopExecuting(this.keepAliveTimer);
         this.reset();
         if (willReconnect) {
-            this.connect().catch(e => this.$error.next(e));
+            this.transport.reset();
+            this.transformer = this.createTransformer();
+            this.transformer.options.debug = (_a = this.transformer.options.debug) !== null && _a !== void 0 ? _a : this.mqttDebug.extend('transformer');
+            this.connect().catch(e => this.emitError(e));
         }
@@ -349,0 +324,0 @@ }

dist/mqtt.constants.d.ts

@@ -1,16 +0,34 @@
-export declare const PacketTypes: {
-    TYPE_CONNECT: number;
-    TYPE_CONNACK: number;
-    TYPE_PUBLISH: number;
-    TYPE_PUBACK: number;
-    TYPE_PUBREC: number;
-    TYPE_PUBREL: number;
-    TYPE_PUBCOMP: number;
-    TYPE_SUBSCRIBE: number;
-    TYPE_SUBACK: number;
-    TYPE_UNSUBSCRIBE: number;
-    TYPE_UNSUBACK: number;
-    TYPE_PINGREQ: number;
-    TYPE_PINGRESP: number;
-    TYPE_DISCONNECT: number;
-};
+export declare enum PacketType {
+    Connect = 1,
+    ConnAck = 2,
+    Publish = 3,
+    PubAck = 4,
+    PubRec = 5,
+    PubRel = 6,
+    PubComp = 7,
+    Subscribe = 8,
+    SubAck = 9,
+    Unsubscribe = 10,
+    UnsubAck = 11,
+    PingReq = 12,
+    PingResp = 13,
+    Disconnect = 14
+}
+export interface EventMapping {
+    CONNECT: PacketType.Connect;
+    CONNACK: PacketType.ConnAck;
+    PUBLISH: PacketType.Publish;
+    PUBACK: PacketType.PubAck;
+    PUBREC: PacketType.PubRec;
+    PUBREL: PacketType.PubRel;
+    PUBCOMP: PacketType.PubComp;
+    SUBSCRIBE: PacketType.Subscribe;
+    SUBACK: PacketType.SubAck;
+    UNSUBSCRIBE: PacketType.Unsubscribe;
+    UNSUBACK: PacketType.UnsubAck;
+    PINGREQ: PacketType.PingReq;
+    PINGRESP: PacketType.PingResp;
+    DISCONNECT: PacketType.Disconnect;
+}
+export declare type PacketName = keyof EventMapping;
+export declare function packetTypeToString(type: PacketType): PacketName;

dist/mqtt.constants.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-exports.PacketTypes = {
-    TYPE_CONNECT: 1,
-    TYPE_CONNACK: 2,
-    TYPE_PUBLISH: 3,
-    TYPE_PUBACK: 4,
-    TYPE_PUBREC: 5,
-    TYPE_PUBREL: 6,
-    TYPE_PUBCOMP: 7,
-    TYPE_SUBSCRIBE: 8,
-    TYPE_SUBACK: 9,
-    TYPE_UNSUBSCRIBE: 10,
-    TYPE_UNSUBACK: 11,
-    TYPE_PINGREQ: 12,
-    TYPE_PINGRESP: 13,
-    TYPE_DISCONNECT: 14,
-};
+exports.packetTypeToString = exports.PacketType = void 0;
+var PacketType;
+(function (PacketType) {
+    PacketType[PacketType["Connect"] = 1] = "Connect";
+    PacketType[PacketType["ConnAck"] = 2] = "ConnAck";
+    PacketType[PacketType["Publish"] = 3] = "Publish";
+    PacketType[PacketType["PubAck"] = 4] = "PubAck";
+    PacketType[PacketType["PubRec"] = 5] = "PubRec";
+    PacketType[PacketType["PubRel"] = 6] = "PubRel";
+    PacketType[PacketType["PubComp"] = 7] = "PubComp";
+    PacketType[PacketType["Subscribe"] = 8] = "Subscribe";
+    PacketType[PacketType["SubAck"] = 9] = "SubAck";
+    PacketType[PacketType["Unsubscribe"] = 10] = "Unsubscribe";
+    PacketType[PacketType["UnsubAck"] = 11] = "UnsubAck";
+    PacketType[PacketType["PingReq"] = 12] = "PingReq";
+    PacketType[PacketType["PingResp"] = 13] = "PingResp";
+    PacketType[PacketType["Disconnect"] = 14] = "Disconnect";
+})(PacketType = exports.PacketType || (exports.PacketType = {}));
+const reverseMapping = Object.fromEntries(Object.entries(PacketType).map(([k, v]) => [v, k]));
+function packetTypeToString(type) {
+    return reverseMapping[type].toUpperCase();
+}
+exports.packetTypeToString = packetTypeToString;
 //# sourceMappingURL=mqtt.constants.js.map

dist/mqtt.packet.d.ts

@@ -1,26 +0,9 @@
-import { PacketStream } from './packet-stream';
-export declare abstract class MqttPacket {
-    get packetType(): number;
-    set identifier(value: number | null);
-    get identifier(): number | null;
-    protected get hasIdentifier(): boolean;
-    protected get inlineIdentifier(): boolean;
-    private readonly _packetType;
-    protected packetFlags: number;
-    remainingPacketLength: number;
-    private _identifier;
-    private static nextId;
-    protected generateIdentifier(): number;
-    static generateIdentifier(): number;
-    protected constructor(packetType: number);
-    read(stream: PacketStream): void;
-    protected readIdentifier(stream: PacketStream): PacketStream;
-    write(stream: PacketStream): void;
-    protected writeIdentifier(stream: PacketStream): PacketStream;
-    private readRemainingLength;
-    private writeRemainingLength;
-    protected assertValidStringLength(str: string): void;
-    protected assertValidString(str: string): void;
-    protected assertValidQosLevel(level: number): void;
-    toString(): string;
+export declare class IdentifierPacket {
+    readonly identifier: number;
+    constructor(identifier: number);
 }
+export interface PacketWriteResult {
+    flags?: number;
+    identifier?: number;
+}
+export declare function generateIdentifier(): number;

dist/mqtt.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const packet_stream_1 = require("./packet-stream");
-const mqtt_utilities_1 = require("./mqtt.utilities");
-class MqttPacket {
-    constructor(packetType) {
-        this.packetFlags = 0;
-        this.remainingPacketLength = 0;
-        this._packetType = packetType;
+exports.generateIdentifier = exports.IdentifierPacket = void 0;
+class IdentifierPacket {
+    constructor(identifier) {
+        this.identifier = identifier;
     }
-    get packetType() {
-        return this._packetType;
-    }
-    set identifier(value) {
-        if (this.hasIdentifier) {
-            this._identifier = Math.max(Math.min((value !== null && value !== void 0 ? value : 0), 0xffff), 0);
-        }
-    }
-    get identifier() {
-        return this.hasIdentifier ? this._identifier : null;
-    }
-    get hasIdentifier() {
-        return false;
-    }
-    get inlineIdentifier() {
-        return false;
-    }
-    generateIdentifier() {
-        if (mqtt_utilities_1.nullOrUndefined(this._identifier)) {
-            this._identifier = MqttPacket.generateIdentifier();
-        }
-        return this._identifier;
-    }
-    static generateIdentifier() {
-        MqttPacket.nextId++;
-        MqttPacket.nextId &= 0xffff;
-        return MqttPacket.nextId;
-    }
-    read(stream) {
-        const typeAndFlags = stream.readByte();
-        const type = (typeAndFlags & 0xf0) >> 4;
-        const flags = typeAndFlags & 0x0f;
-        if (type !== this._packetType) {
-            throw new Error('Invalid packet type');
-        }
-        this.packetFlags = flags;
-        this.readRemainingLength(stream);
-        if (this.hasIdentifier && !this.inlineIdentifier) {
-            this.readIdentifier(stream);
-        }
-    }
-    readIdentifier(stream) {
-        if (this.hasIdentifier)
-            this._identifier = stream.readWord();
-        return stream;
-    }
-    write(stream) {
-        stream.writeByte(((this._packetType & 0x0f) << 4) | (this.packetFlags & 0x0f));
-        if (this.hasIdentifier && !this.inlineIdentifier)
-            this.remainingPacketLength += 2;
-        this.writeRemainingLength(stream);
-        if (this.hasIdentifier && !this.inlineIdentifier)
-            this.writeIdentifier(stream);
-    }
-    writeIdentifier(stream) {
-        var _a;
-        if (this.hasIdentifier)
-            stream.writeWord((_a = this._identifier, (_a !== null && _a !== void 0 ? _a : this.generateIdentifier())));
-        return stream;
-    }
-    readRemainingLength(stream) {
-        this.remainingPacketLength = 0;
-        let multiplier = 1;
-        let encodedByte;
-        do {
-            encodedByte = stream.readByte();
-            this.remainingPacketLength += (encodedByte & 0x7f) * multiplier;
-            if (multiplier > 128 * 128 * 128) {
-                throw new Error(`Invalid length @${stream.position}/${stream.length}; currentLength: ${this.remainingPacketLength}`);
-            }
-            multiplier *= 0x80;
-        } while ((encodedByte & 0x80) !== 0);
-    }
-    writeRemainingLength(stream) {
-        let num = this.remainingPacketLength;
-        let digit = 0;
-        do {
-            digit = num % 128 | 0;
-            num = (num / 128) | 0;
-            if (num > 0)
-                digit = digit | 0x80;
-            stream.writeByte(digit);
-        } while (num > 0);
-    }
-    assertValidStringLength(str) {
-        if (str.length > 0xffff) {
-            throw new Error(`The string ${str.substring(0, 20)} is longer than 0xffff bytes.`);
-        }
-    }
-    assertValidString(str) {
-        this.assertValidStringLength(str);
-        /* eslint no-control-regex: "off" */
-        if (str.match(/[\xD8-\xDF][\x00-\xFF]|\x00\x00/) !== null) {
-            throw new Error(`The string ${str.substring(0, 20)} contains invalid characters`);
-        }
-    }
-    assertValidQosLevel(level) {
-        if (level < 0 || level > 2) {
-            throw new Error(`Invalid QoS level ${level}.`);
-        }
-    }
-    toString() {
-        const stream = packet_stream_1.PacketStream.empty();
-        return stream.data.toString('utf8');
-    }
 }
-exports.MqttPacket = MqttPacket;
-MqttPacket.nextId = 0;
+exports.IdentifierPacket = IdentifierPacket;
+let _lastId = 0;
+function generateIdentifier() {
+    return (_lastId = (_lastId + 1) & 0xffff);
+}
+exports.generateIdentifier = generateIdentifier;
 //# sourceMappingURL=mqtt.packet.js.map

dist/mqtt.parser.d.ts

 /// <reference types="node" />
-import { MqttPacket } from './mqtt.packet';
-import { PacketStream } from './packet-stream';
-export declare class MqttParser {
-    protected debug?: ((msg: string) => void) | undefined;
-    protected stream: PacketStream;
-    protected errorCallback: (e: Error) => void;
-    protected lock: import("./mqtt.utilities").Lock;
-    mapping: [number, () => MqttPacket][];
-    constructor(errorCallback?: (e: Error) => void, debug?: ((msg: string) => void) | undefined);
+import { PacketType } from './mqtt.constants';
+import { Transform, TransformCallback } from 'stream';
+import { Debugger } from 'debug';
+import { DefaultPacketReadResultMap, PacketReadMap, PacketReadResultMap } from './packets/packet-reader';
+export interface MqttParseResult<ReadMap extends PacketReadResultMap, T extends PacketType> {
+    type: T;
+    flags: number;
+    data: ReadMap[T];
+}
+export interface MqttTransformerOptions<T extends PacketReadResultMap = DefaultPacketReadResultMap> {
+    debug?: Debugger;
+    mapping?: PacketReadMap<T>;
+}
+export declare class MqttTransformer<ReadMap extends PacketReadResultMap = DefaultPacketReadResultMap> extends Transform {
+    options: MqttTransformerOptions<ReadMap>;
+    mapping: PacketReadMap<ReadMap>;
+    private internalStream;
+    constructor(options?: MqttTransformerOptions<ReadMap>);
+    _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void;
     reset(): void;
-    parse(data: Buffer): Promise<MqttPacket[]>;
 }

dist/mqtt.parser.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("./mqtt.constants");
+exports.MqttTransformer = void 0;
 const packet_stream_1 = require("./packet-stream");
 const errors_1 = require("./errors");
-const packets_1 = require("./packets");
-const mqtt_utilities_1 = require("./mqtt.utilities");
-class MqttParser {
-    constructor(errorCallback, debug) {
-        this.debug = debug;
-        this.lock = mqtt_utilities_1.createLock();
-        this.mapping = [
-            [mqtt_constants_1.PacketTypes.TYPE_CONNACK, () => new packets_1.ConnectResponsePacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PUBLISH, () => new packets_1.PublishRequestPacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PUBACK, () => new packets_1.PublishAckPacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PUBREC, () => new packets_1.PublishReceivedPacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PUBREL, () => new packets_1.PublishReleasePacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PUBCOMP, () => new packets_1.PublishCompletePacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_SUBSCRIBE, () => new packets_1.SubscribeRequestPacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_SUBACK, () => new packets_1.SubscribeResponsePacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_UNSUBSCRIBE, () => new packets_1.UnsubscribeRequestPacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_UNSUBACK, () => new packets_1.UnsubscribeResponsePacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PINGREQ, () => new packets_1.PingRequestPacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_PINGRESP, () => new packets_1.PingResponsePacket()],
-            [mqtt_constants_1.PacketTypes.TYPE_DISCONNECT, () => new packets_1.DisconnectRequestPacket()],
-        ];
-        this.stream = packet_stream_1.PacketStream.empty();
-        /* eslint @typescript-eslint/no-empty-function: "off" */
-        this.errorCallback = (errorCallback !== null && errorCallback !== void 0 ? errorCallback : (() => { }));
+const stream_1 = require("stream");
+const packet_reader_1 = require("./packets/packet-reader");
+class MqttTransformer extends stream_1.Transform {
+    constructor(options = {}) {
+        super({ objectMode: true });
+        this.options = options;
+        // force the type here
+        this.mapping = packet_reader_1.DefaultPacketReadMap;
+        this.internalStream = undefined;
+        this.mapping = {
+            ...this.mapping,
+            ...options.mapping,
+        };
     }
-    reset() {
-        this.stream = packet_stream_1.PacketStream.empty();
-        this.lock.locked = false;
-        this.lock.resolver = null;
-    }
-    async parse(data) {
-        var _a, _b, _c, _d, _e, _f, _g, _h;
-        await this.lock.wait();
-        this.lock.lock();
-        let startPos = this.stream.position;
-        this.stream.write(data);
-        this.stream.position = startPos;
-        const results = [];
-        try {
-            while (this.stream.remainingBytes > 0) {
-                const type = this.stream.readByte() >> 4;
-                let packet;
-                try {
-                    // @ts-ignore - if undefined -> catched
-                    packet = this.mapping.find(x => x[0] === type)[1]();
+    _transform(chunk, encoding, callback) {
+        var _a, _b;
+        if (!Buffer.isBuffer(chunk)) {
+            callback(new Error('Expected a Buffer'));
+            return;
+        }
+        const stream = this.internalStream ? this.internalStream.write(chunk, false) : packet_stream_1.PacketStream.fromBuffer(chunk);
+        this.internalStream = undefined;
+        let startPos = 0;
+        while (stream.remainingBytes > 0) {
+            const firstByte = stream.readByte();
+            const type = (firstByte >> 4);
+            const flags = (firstByte & 0x0f);
+            const packetFn = this.mapping[type];
+            if (!packetFn) {
+                callback(new errors_1.UnexpectedPacketError(`No packet found for ${type}; @${stream.position}/${stream.length}`));
+                return;
+            }
+            let remainingLength = -1;
+            try {
+                remainingLength = stream.readVariableByteInteger();
+                const packet = packetFn(stream, remainingLength, flags);
+                this.push({
+                    type,
+                    data: packet,
+                    flags
+                });
+                stream.cut();
+                startPos = stream.position;
+            }
+            catch (e) {
+                if (e instanceof errors_1.EndOfStreamError) {
+                    (_b = (_a = this.options).debug) === null || _b === void 0 ? void 0 : _b.call(_a, `EOS:\n  ${remainingLength} got: ${stream.length} (+) ${chunk.byteLength};\n  return: ${startPos};`);
+                    stream.position = startPos;
+                    this.internalStream = stream.cut();
+                    callback();
+                    return;
                 }
-                catch (e) {
-                    (_b = (_a = this).debug) === null || _b === void 0 ? void 0 : _b.call(_a, `No packet found for ${type};
-                         @${this.stream.position}/${this.stream.length}
-                         parsed: ${results.length}`);
-                    continue;
-                }
-                this.stream.seek(-1);
-                let exitParser = false;
-                try {
-                    packet.read(this.stream);
-                    results.push(packet);
-                    this.stream.cut();
-                    startPos = this.stream.position;
-                }
-                catch (e) {
-                    if (e instanceof errors_1.EndOfStreamError) {
-                        (_d = (_c = this).debug) === null || _d === void 0 ? void 0 : _d.call(_c, `EOS:\n  ${packet.remainingPacketLength} got: ${this.stream.length} (+) ${data.byteLength};\n  return: ${startPos};`);
-                        this.stream.position = startPos;
-                        exitParser = true;
-                    }
-                    else {
-                        (_f = (_e = this).debug) === null || _f === void 0 ? void 0 : _f.call(_e, `Error in parser (type: ${type}): 
+                else {
+                    callback(new errors_1.MalformedPacketError(`Error in parser (type: ${type}): 
                         ${e.stack}; 
                         exiting; 
                         resetting;
-                        stream: ${this.stream.data.toString('base64')}`);
-                        this.errorCallback(e);
-                        exitParser = true;
-                        this.stream = packet_stream_1.PacketStream.empty();
-                    }
+                        stream: ${stream.data.toString('base64')}`));
+                    return;
                 }
-                if (exitParser)
-                    break;
             }
         }
-        catch (e) {
-            (_h = (_g = this).debug) === null || _h === void 0 ? void 0 : _h.call(_g, `Error in parser: 
-                ${e.stack};
-                 resetting; 
-                 stream: ${this.stream.data.toString('base64')}`);
-            this.stream = packet_stream_1.PacketStream.empty();
-            this.errorCallback(e);
-        }
-        this.lock.unlock();
-        return results;
+        callback();
     }
+    reset() {
+        this.internalStream = undefined;
+    }
 }
-exports.MqttParser = MqttParser;
+exports.MqttTransformer = MqttTransformer;
 //# sourceMappingURL=mqtt.parser.js.map

dist/mqtt.types.d.ts

 import { ConnectRequestOptions } from './packets';
-import { MqttParser } from './mqtt.parser';
+import { MqttTransformer } from './mqtt.parser';
 import { Transport } from './transport';
 import { XOR } from 'ts-xor';
 import { MqttMessage } from './mqtt.message';
-export declare type MqttClientConstructorOptions = XOR<{
+import { DefaultPacketReadResultMap, PacketReadMap, PacketReadResultMap } from './packets/packet-reader';
+import { DefaultPacketWriteOptions, PacketWriteMap, PacketWriteOptionsMap, PacketWriter } from './packets/packet-writer';
+import { TransformerFn, ValidatorFn } from './mqtt.listener';
+export declare type MqttClientConstructorOptions<ReadMap extends PacketReadResultMap = DefaultPacketReadResultMap, WriteMap extends PacketWriteOptionsMap = DefaultPacketWriteOptions> = XOR<{
     transport: Transport<unknown>;
 }, {
-    url: string;
+    host: string;
+    port: number;
     enableTrace?: boolean;
 }> & {
-    parser?: MqttParser;
+    readMap?: PacketReadMap<ReadMap>;
+    createTransformer?: () => MqttTransformer<ReadMap>;
+    writeMap?: PacketWriteMap<WriteMap>;
+    packetWriter?: PacketWriter<WriteMap>;
     autoReconnect?: boolean;
@@ -20,5 +27,6 @@ };
 export declare type RegisterClientOptions = ConnectRequestOptions;
+export declare type TimerRef = any;
 export declare type ExecuteNextTick = (action: () => void) => void;
-export declare type ExecutePeriodically = (timeInMs: number, action: () => void) => object;
-export declare type ExecuteDelayed = (timeInMs: number, action: () => void) => object;
+export declare type ExecutePeriodically = (timeInMs: number, action: () => void) => TimerRef;
+export declare type ExecuteDelayed = (timeInMs: number, action: () => void) => TimerRef;
 export declare type StopExecuting = (ref: any) => void;
@@ -32,13 +40,16 @@ export declare type AsyncLike<TIn, TOut> = (data: TIn) => TOut | PromiseLike<TOut>;
 }
-export interface ListenOptions<TOut> {
+export interface ListenOptions<TOut, Params extends Record<string, string>> {
     topic: string;
-    validator?: null | ((data: MqttMessage) => boolean);
-    transformer?: (data: IncomingListenMessage<any>) => TOut;
+    validator?: ValidatorFn<Params>;
+    transformer?: TransformerFn<TOut, Params>;
 }
-export interface ListenSubscribeOptions<TOut> extends ListenOptions<TOut> {
+export interface ListenSubscribeOptions<TOut, Params extends Record<string, string>> extends ListenOptions<TOut, Params> {
     subscriptionInfo?: Partial<MqttSubscription>;
 }
-export interface IncomingListenMessage<T> extends MqttMessage {
+export interface IncomingListenMessage<T extends Record<string, string> = Record<string, string>> extends MqttMessage {
     params?: T;
 }
-export declare type Resolvable<T extends object> = (() => Promise<T>) | (() => T) | T;
+export declare type Resolvable<T extends Record<string, unknown>> = (() => Promise<T>) | (() => T) | T;
+export interface IdentifierData {
+    identifier: number;
+}

dist/mqtt.utilities.d.ts

@@ -1,14 +0,9 @@
-import { ListenerInfo, Resolvable } from './mqtt.types';
-import { MqttMessage } from './mqtt.message';
-import { MqttPacket } from './mqtt.packet';
-import { ConnectRequestPacket, ConnectResponsePacket, DisconnectRequestPacket, PingRequestPacket, PingResponsePacket, PublishAckPacket, PublishCompletePacket, PublishReceivedPacket, PublishReleasePacket, PublishRequestPacket, SubscribeRequestPacket, SubscribeResponsePacket, UnsubscribeRequestPacket, UnsubscribeResponsePacket } from './packets';
-export declare function topicListener<T>(options: {
-    topic: string;
-    transformer: (data: MqttMessage) => T | PromiseLike<T>;
-    validator?: (data: MqttMessage) => boolean | PromiseLike<boolean>;
-    onData: (data: T) => void | PromiseLike<void>;
-}): ListenerInfo<MqttMessage, T>;
+/// <reference types="node" />
+import { Resolvable } from './mqtt.types';
+import { ConnectResponsePacket, PingResponsePacket, PublishAckPacket, PublishCompletePacket, PublishReceivedPacket, PublishReleasePacket, PublishRequestPacket, SubscribeResponsePacket, UnsubscribeResponsePacket } from './packets';
+import { PacketType } from './mqtt.constants';
 export declare function matchTopic(baseTopic: string, incomingTopic: string): boolean;
+export declare function expectRemainingLength(length: number, expected?: number): void;
 export declare function removeUntil(input: string, char: string): string;
-export declare function extractParams(template: string, topic: string): object;
+export declare function extractParams(template: string, topic: string): Record<string, string>;
 export interface Resolvers<T> {
@@ -18,32 +13,17 @@ resolve: (value: T) => void;
 }
-export declare const nullOrUndefined: (input: any) => boolean;
-export declare function isPacket(target: any, type: number): boolean;
-export declare const isConnect: (target: MqttPacket) => target is ConnectRequestPacket;
-export declare const isConnAck: (target: MqttPacket) => target is ConnectResponsePacket;
-export declare const isPublish: (target: MqttPacket) => target is PublishRequestPacket;
-export declare const isPubAck: (target: MqttPacket) => target is PublishAckPacket;
-export declare const isPubRec: (target: MqttPacket) => target is PublishReceivedPacket;
-export declare const isPubRel: (target: MqttPacket) => target is PublishReleasePacket;
-export declare const isPubComp: (target: MqttPacket) => target is PublishCompletePacket;
-export declare const isSubscribe: (target: MqttPacket) => target is SubscribeRequestPacket;
-export declare const isSubAck: (target: MqttPacket) => target is SubscribeResponsePacket;
-export declare const isUnsubscribe: (target: MqttPacket) => target is UnsubscribeRequestPacket;
-export declare const isUnsubAck: (target: MqttPacket) => target is UnsubscribeResponsePacket;
-export declare const isPingReq: (target: MqttPacket) => target is PingRequestPacket;
-export declare const isPingResp: (target: MqttPacket) => target is PingResponsePacket;
-export declare const isDisconnect: (target: MqttPacket) => target is DisconnectRequestPacket;
-/**
- * Some workaround for async requests:
- * This prevents the execution if there's already something in the buffer.
- * Note: if something fails, this will lock forever
- * @type {{unlock: () => void; resolve: null; lock: () => void; locked: boolean}}
- */
-export declare function createLock(): Lock;
-export declare function resolve<T extends object>(resolvable: Resolvable<T>): Promise<T>;
-export interface Lock {
-    resolver: Function | null;
-    locked: boolean;
-    lock: () => void;
-    unlock: () => void;
-    wait: () => Promise<void>;
-}
+export declare const nullOrUndefined: (input: unknown) => boolean;
+export declare const isConnAck: (target: unknown) => target is ConnectResponsePacket;
+export declare const isPublish: (target: unknown) => target is PublishRequestPacket;
+export declare const isPubAck: (target: unknown) => target is PublishAckPacket;
+export declare const isPubRec: (target: unknown) => target is PublishReceivedPacket;
+export declare const isPubRel: (target: unknown) => target is PublishReleasePacket;
+export declare const isPubComp: (target: unknown) => target is PublishCompletePacket;
+export declare const isSubAck: (target: unknown) => target is SubscribeResponsePacket;
+export declare const isUnsubAck: (target: unknown) => target is UnsubscribeResponsePacket;
+export declare const isPingResp: (target: unknown) => target is PingResponsePacket;
+export declare function resolve<T extends Record<string, unknown>>(resolvable: Resolvable<T>): Promise<T>;
+export declare function notUndefined<T>(value: T | undefined): value is T;
+export declare function toBuffer(value: Buffer | string): Buffer;
+export declare function toMqttTopicFilter(paramString: string): [string, string?];
+export declare function createDefaultPacketLogger(debug: (data: string) => void): (packetType: PacketType, packetInfo: Record<string, string | number | boolean | undefined>) => void;
+export declare function stringifyObject(obj: Record<string, unknown>): string;

dist/mqtt.utilities.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.stringifyObject = exports.createDefaultPacketLogger = exports.toMqttTopicFilter = exports.toBuffer = exports.notUndefined = exports.resolve = exports.isPingResp = exports.isUnsubAck = exports.isSubAck = exports.isPubComp = exports.isPubRel = exports.isPubRec = exports.isPubAck = exports.isPublish = exports.isConnAck = exports.nullOrUndefined = exports.extractParams = exports.removeUntil = exports.expectRemainingLength = exports.matchTopic = void 0;
+const packets_1 = require("./packets");
 const mqtt_constants_1 = require("./mqtt.constants");
-function topicListener(options) {
-    return {
-        eventName: 'message',
-        validator: data => {
-            if (data.topic === options.topic) {
-                return options.validator ? options.validator(data) : true;
-            }
-            return false;
-        },
-        transformer: options.transformer,
-        onData: options.onData,
-    };
-}
-exports.topicListener = topicListener;
 function matchTopic(baseTopic, incomingTopic) {
@@ -32,2 +20,11 @@ if (baseTopic.length === incomingTopic.length && baseTopic === incomingTopic)
 exports.matchTopic = matchTopic;
+function expectRemainingLength(length, expected) {
+    if (!expected) {
+        expected = 0;
+    }
+    if (length !== expected) {
+        throw new Error(`Expected remaining length to be ${expected} but got ${length}`);
+    }
+}
+exports.expectRemainingLength = expectRemainingLength;
 function removeUntil(input, char) {
@@ -50,57 +47,68 @@ return input.substring(Math.max(input.indexOf(char), 0));
 exports.nullOrUndefined = (input) => input == undefined;
-function isPacket(target, type) {
-    return target.packetType === type;
+exports.isConnAck = (target) => target instanceof packets_1.ConnectResponsePacket;
+exports.isPublish = (target) => target instanceof packets_1.PublishRequestPacket;
+exports.isPubAck = (target) => target instanceof packets_1.PublishAckPacket;
+exports.isPubRec = (target) => target instanceof packets_1.PublishReceivedPacket;
+exports.isPubRel = (target) => target instanceof packets_1.PublishReleasePacket;
+exports.isPubComp = (target) => target instanceof packets_1.PublishCompletePacket;
+exports.isSubAck = (target) => target instanceof packets_1.SubscribeResponsePacket;
+exports.isUnsubAck = (target) => target instanceof packets_1.UnsubscribeResponsePacket;
+exports.isPingResp = (target) => target instanceof packets_1.PingResponsePacket;
+async function resolve(resolvable) {
+    return typeof resolvable === 'object' ? resolvable : await resolvable();
 }
-exports.isPacket = isPacket;
-exports.isConnect = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_CONNECT);
-exports.isConnAck = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_CONNACK);
-exports.isPublish = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PUBLISH);
-exports.isPubAck = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PUBACK);
-exports.isPubRec = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PUBREC);
-exports.isPubRel = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PUBREL);
-exports.isPubComp = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PUBCOMP);
-exports.isSubscribe = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_SUBSCRIBE);
-exports.isSubAck = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_SUBACK);
-exports.isUnsubscribe = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_UNSUBSCRIBE);
-exports.isUnsubAck = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_UNSUBACK);
-exports.isPingReq = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PINGREQ);
-exports.isPingResp = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_PINGRESP);
-exports.isDisconnect = (target) => isPacket(target, mqtt_constants_1.PacketTypes.TYPE_DISCONNECT);
-/**
- * Some workaround for async requests:
- * This prevents the execution if there's already something in the buffer.
- * Note: if something fails, this will lock forever
- * @type {{unlock: () => void; resolve: null; lock: () => void; locked: boolean}}
- */
-function createLock() {
-    return {
-        locked: false,
-        lock() {
-            this.locked = true;
-        },
-        unlock() {
-            this.locked = false;
-            if (this.resolver) {
-                this.resolver();
-                this.resolver = null;
-            }
-        },
-        resolver: null,
-        wait() {
-            if (this.locked) {
-                return new Promise(resolve => {
-                    this.resolver = resolve;
-                });
-            }
-            else {
-                return Promise.resolve();
-            }
-        },
+exports.resolve = resolve;
+function notUndefined(value) {
+    return typeof value !== 'undefined';
+}
+exports.notUndefined = notUndefined;
+function toBuffer(value) {
+    return typeof value === 'string' ? Buffer.from(value) : value;
+}
+exports.toBuffer = toBuffer;
+const paramRegex = /\/:[A-Za-z-_0-9]+/g;
+function toMqttTopicFilter(paramString) {
+    if (paramString.match(paramRegex)) {
+        return [paramString.replace(paramRegex, '/+'), paramString];
+    }
+    return [paramString];
+}
+exports.toMqttTopicFilter = toMqttTopicFilter;
+function createDefaultPacketLogger(debug) {
+    return (packetType, packetInfo) => {
+        if (packetType !== mqtt_constants_1.PacketType.PingReq && packetType !== mqtt_constants_1.PacketType.PingResp) {
+            debug(`Write ${mqtt_constants_1.packetTypeToString(packetType)} ${stringifyObject(packetInfo)}`);
+        }
     };
 }
-exports.createLock = createLock;
-async function resolve(resolvable) {
-    return typeof resolvable === 'object' ? resolvable : await resolvable();
+exports.createDefaultPacketLogger = createDefaultPacketLogger;
+function stringifyObject(obj) {
+    return `${obj.constructor === Object ? '' : `${obj.constructor.name} `}{${Object.entries(obj)
+        .filter(([, v]) => typeof v !== 'undefined')
+        .map(([k, v]) => `${k}: ${stringifyValue(v)}`)
+        .join(', ')}}`;
 }
-exports.resolve = resolve;
+exports.stringifyObject = stringifyObject;
+function stringifyValue(value) {
+    if (typeof value === 'object') {
+        if (value === null) {
+            return '<null>';
+        }
+        else if (Array.isArray(value)) {
+            return `<Array { len: ${value.length}}>`;
+        }
+        else if (Buffer.isBuffer(value)) {
+            return `<Buffer { bytes: ${value.byteLength}}>`;
+        }
+        else if (value.constructor !== Object) {
+            return `<${value.constructor.name}>`;
+        }
+        else {
+            return '{...}';
+        }
+    }
+    else {
+        return typeof value === 'string' ? `"${value}"` : String(value);
+    }
+}
 //# sourceMappingURL=mqtt.utilities.js.map

dist/packet-stream.d.ts

@@ -15,2 +15,3 @@ /// <reference types="node" />
     static empty(): PacketStream;
+    static fromHex(hex: string): PacketStream;
     /**
@@ -24,3 +25,3 @@ *
     cut(): this;
-    write(data: Buffer): this;
+    write(data: Buffer, move?: boolean): this;
     writeRawString(data: string): this;
@@ -30,2 +31,4 @@ writeByte(num: number): this;
     writeString(str: string): this;
+    writeRawAndLength(data: Buffer): this;
+    writeVariableByteInteger(value: number): this;
     read(len: number): Buffer;
@@ -37,2 +40,3 @@ readSlice(end: number): Buffer;
     readStringAsBuffer(): Buffer;
+    readVariableByteInteger(): number;
 }

dist/packet-stream.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.PacketStream = void 0;
 /*
@@ -12,2 +13,3 @@
     constructor(data, length, buffer) {
+        this._position = 0;
         this._data = data ? Buffer.from(data) : length ? Buffer.alloc(length) : buffer ? buffer : Buffer.from([]);
@@ -43,2 +45,5 @@ this.position = 0;
     }
+    static fromHex(hex) {
+        return PacketStream.fromBuffer(Buffer.from(hex, 'hex'));
+    }
     /**
@@ -66,3 +71,3 @@ *
     // Write
-    write(data) {
+    write(data, move = true) {
         if (this._data)
@@ -72,3 +77,4 @@ this._data = Buffer.concat([this._data, data]);
             this._data = data;
-        this.move(data.length);
+        if (move)
+            this.move(data.length);
         return this;
@@ -90,2 +96,17 @@ }
     }
+    writeRawAndLength(data) {
+        this.writeWord(data.byteLength);
+        return this.write(data);
+    }
+    writeVariableByteInteger(value) {
+        let digit = 0;
+        do {
+            digit = value % 128 | 0;
+            value = (value / 128) | 0;
+            if (value > 0)
+                digit = digit | 0x80;
+            this.writeByte(digit);
+        } while (value > 0);
+        return this;
+    }
     // Read
@@ -118,4 +139,18 @@ read(len) {
     }
+    readVariableByteInteger() {
+        let value = 0;
+        let multiplier = 1;
+        let encodedByte;
+        do {
+            encodedByte = this.readByte();
+            value += (encodedByte & 0x7f) * multiplier;
+            if (multiplier > 128 * 128 * 128) {
+                throw new Error(`Invalid variable byte integer ${this.position}/${this.length}; currentValue: ${value}`);
+            }
+            multiplier *= 0x80;
+        } while ((encodedByte & 0x80) !== 0);
+        return value;
+    }
 }
 exports.PacketStream = PacketStream;
 //# sourceMappingURL=packet-stream.js.map

dist/packets/connect.request.packet.d.ts

@@ -1,22 +0,22 @@
-import { MqttPacket } from '../mqtt.packet';
+/// <reference types="node" />
+import { PacketWriteResult } from '../mqtt.packet';
 import { PacketStream } from '../packet-stream';
-import { MqttMessage } from '../mqtt.message';
-export declare class ConnectRequestPacket extends MqttPacket {
-    options: ConnectRequestOptions;
-    constructor(options?: ConnectRequestOptions);
-    private static makeFlags;
-    write(stream: PacketStream): void;
-    read(): void;
-}
-export interface ConnectRequestOptions {
-    protocolLevel?: number;
-    protocolName?: string;
-    flags?: number;
-    clientId?: string;
-    keepAlive?: number;
-    will?: MqttMessage;
+export declare type ConnectRequestOptions = Partial<RequiredConnectRequestOptions>;
+export interface RequiredConnectRequestOptions {
+    protocolLevel: number;
+    protocolName: string;
+    clientId: string;
+    keepAlive: number;
+    will?: {
+        topic: string;
+        message: Buffer | string;
+        retained?: boolean;
+        qosLevel?: number;
+    };
     username?: string;
-    password?: string;
-    clean?: boolean;
-    connectDelay?: number | null;
+    password?: Buffer | string;
+    clean: boolean;
+    connectDelay?: number;
 }
+export declare function writeConnectPacket(stream: PacketStream, options: RequiredConnectRequestOptions): PacketWriteResult;
+export declare function makeFlags(options: ConnectRequestOptions): number;

dist/packets/connect.request.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_packet_1 = require("../mqtt.packet");
-const mqtt_constants_1 = require("../mqtt.constants");
-const packet_stream_1 = require("../packet-stream");
-const lodash_1 = require("lodash");
+exports.makeFlags = exports.writeConnectPacket = void 0;
 const errors_1 = require("../errors");
-class ConnectRequestPacket extends mqtt_packet_1.MqttPacket {
-    constructor(options) {
-        super(mqtt_constants_1.PacketTypes.TYPE_CONNECT);
-        this.options = lodash_1.defaults(options, {
-            protocolLevel: 4,
-            protocolName: 'MQTT',
-            flags: ConnectRequestPacket.makeFlags(options),
-            clientId: 'mqtts_' + lodash_1.random(0, 200000),
-            keepAlive: 60,
-        });
+const mqtt_utilities_1 = require("../mqtt.utilities");
+function writeConnectPacket(stream, options) {
+    // Variable Header
+    stream
+        .writeString(options.protocolName)
+        .writeByte(options.protocolLevel)
+        .writeByte(makeFlags(options))
+        .writeWord(options.keepAlive);
+    // Payload
+    stream.writeString(options.clientId);
+    options.will && stream
+        .writeString(options.will.topic)
+        .writeRawAndLength(mqtt_utilities_1.toBuffer(options.will.message));
+    options.username && stream.writeString(options.username);
+    options.password && stream.writeRawAndLength(mqtt_utilities_1.toBuffer(options.password));
+    return {};
+}
+exports.writeConnectPacket = writeConnectPacket;
+function makeFlags(options) {
+    var _a;
+    if (!options)
+        return 0;
+    if (mqtt_utilities_1.notUndefined(options.password) && !mqtt_utilities_1.notUndefined(options.username))
+        throw new errors_1.MalformedPacketError('MQTT-3.1.2-22 If the User Name Flag is set to 0, the Password Flag MUST be set to 0');
+    let flags = 0;
+    if (mqtt_utilities_1.notUndefined(options.username))
+        flags |= 0x1 << 7;
+    if (mqtt_utilities_1.notUndefined(options.password))
+        flags |= 0x1 << 6;
+    if (mqtt_utilities_1.notUndefined(options.will)) {
+        if (options.will.retained)
+            flags |= 0x1 << 5;
+        flags |= (((_a = options.will.qosLevel) !== null && _a !== void 0 ? _a : 0) & 0x03) << 3;
+        flags |= 0x1 << 2;
     }
-    static makeFlags(options) {
-        var _a;
-        if (!options)
-            return 0;
-        let flags = 0;
-        if (options.username)
-            flags |= 0x1 << 7;
-        if (options.password)
-            flags |= 0x1 << 6;
-        if (options.will) {
-            if (options.will.retained)
-                flags |= 0x1 << 5;
-            flags |= ((_a = options.will.qosLevel, (_a !== null && _a !== void 0 ? _a : 0)) & 0x03) << 3;
-            flags |= 0x1 << 2;
-        }
-        if (options.clean)
-            flags |= 0x1 << 1;
-        return flags;
-    }
-    write(stream) {
-        const { protocolLevel, protocolName, flags, clientId, keepAlive, will, username, password } = this.options;
-        const data = packet_stream_1.PacketStream.empty()
-            .writeString((protocolName !== null && protocolName !== void 0 ? protocolName : 'MQTT'))
-            .writeByte((protocolLevel !== null && protocolLevel !== void 0 ? protocolLevel : 4))
-            .writeByte((flags !== null && flags !== void 0 ? flags : ConnectRequestPacket.makeFlags(this.options)))
-            .writeWord((keepAlive !== null && keepAlive !== void 0 ? keepAlive : 60))
-            .writeString((clientId !== null && clientId !== void 0 ? clientId : 'mqtt_' + lodash_1.random(0, 200000)));
-        if (will)
-            data.writeString(will.topic).writeString(will.payload.toString());
-        if (username)
-            data.writeString(username);
-        if (password)
-            data.writeString(password);
-        this.remainingPacketLength = data.length;
-        super.write(stream);
-        stream.write(data.data);
-    }
-    read() {
-        throw new errors_1.InvalidDirectionError('read');
-    }
+    if (options.clean)
+        flags |= 0x1 << 1;
+    return flags;
 }
-exports.ConnectRequestPacket = ConnectRequestPacket;
+exports.makeFlags = makeFlags;
 //# sourceMappingURL=connect.request.packet.js.map

dist/packets/connect.response.packet.d.ts

@@ -1,18 +0,18 @@
-/// <reference types="node" />
-import { MqttPacket } from '../mqtt.packet';
 import { PacketStream } from '../packet-stream';
-export declare class ConnectResponsePacket extends MqttPacket {
-    get payload(): Buffer;
-    static readonly returnCodes: string[];
-    get returnCode(): number;
-    get flags(): number;
+export declare class ConnectResponsePacket {
+    ackFlags: number;
+    returnCode: ConnectReturnCode;
+    get sessionPresent(): boolean;
     get isSuccess(): boolean;
-    get isError(): boolean;
-    get errorName(): string;
-    private _flags;
-    private _returnCode;
-    private _payload;
-    constructor();
-    read(stream: PacketStream): void;
-    write(): void;
+    get errorName(): keyof typeof ConnectReturnCode | string;
+    constructor(ackFlags: number, returnCode: ConnectReturnCode);
 }
+export declare function readConnectResponsePacket(stream: PacketStream, remaining: number): ConnectResponsePacket;
+export declare enum ConnectReturnCode {
+    Accepted = 0,
+    UnacceptableProtocolVersion = 1,
+    IdentifierRejected = 2,
+    ServerUnavailable = 3,
+    BadUsernameOrPassword = 4,
+    NotAuthorized = 5
+}

dist/packets/connect.response.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_packet_1 = require("../mqtt.packet");
-const mqtt_constants_1 = require("../mqtt.constants");
-const errors_1 = require("../errors");
-class ConnectResponsePacket extends mqtt_packet_1.MqttPacket {
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_CONNACK);
+exports.ConnectReturnCode = exports.readConnectResponsePacket = exports.ConnectResponsePacket = void 0;
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class ConnectResponsePacket {
+    constructor(ackFlags, returnCode) {
+        this.ackFlags = ackFlags;
+        this.returnCode = returnCode;
     }
-    get payload() {
-        return this._payload;
+    get sessionPresent() {
+        return !!(this.ackFlags & 0x1);
     }
-    get returnCode() {
-        return this._returnCode;
-    }
-    get flags() {
-        return this._flags;
-    }
     get isSuccess() {
-        return this.returnCode === 0;
+        return this.returnCode === ConnectReturnCode.Accepted;
     }
-    get isError() {
-        return this.returnCode > 0;
-    }
     get errorName() {
-        return ConnectResponsePacket.returnCodes[Math.min(this.returnCode, ConnectResponsePacket.returnCodes.length - 1)];
+        var _a, _b;
+        return (_b = (_a = Object.entries(ConnectReturnCode).find(([, v]) => v === this.returnCode)) === null || _a === void 0 ? void 0 : _a[0]) !== null && _b !== void 0 ? _b : 'Unknown';
     }
-    read(stream) {
-        super.read(stream);
-        this._flags = stream.readByte();
-        this._returnCode = stream.readByte();
-        /**
-         *  NOT IN MQTT 3.1.1!
-         */
-        if (this.remainingPacketLength - 2 > 0) {
-            this._payload = stream.readStringAsBuffer();
-        }
+}
+exports.ConnectResponsePacket = ConnectResponsePacket;
+function readConnectResponsePacket(stream, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 2);
+    const ack = stream.readByte();
+    const returnCode = stream.readByte();
+    if (ack > 1) {
+        throw new Error('Invalid ack');
     }
-    write() {
-        throw new errors_1.InvalidDirectionError('write');
+    else if (returnCode > 5) {
+        throw new Error('Invalid return code');
     }
+    return new ConnectResponsePacket(ack, returnCode);
 }
-exports.ConnectResponsePacket = ConnectResponsePacket;
-ConnectResponsePacket.returnCodes = [
-    'Connection accepted',
-    'Unacceptable protocol version',
-    'Identifier rejected',
-    'Server unavailable',
-    'Bad user name or password',
-    'Not authorized',
-];
+exports.readConnectResponsePacket = readConnectResponsePacket;
+var ConnectReturnCode;
+(function (ConnectReturnCode) {
+    ConnectReturnCode[ConnectReturnCode["Accepted"] = 0] = "Accepted";
+    ConnectReturnCode[ConnectReturnCode["UnacceptableProtocolVersion"] = 1] = "UnacceptableProtocolVersion";
+    ConnectReturnCode[ConnectReturnCode["IdentifierRejected"] = 2] = "IdentifierRejected";
+    ConnectReturnCode[ConnectReturnCode["ServerUnavailable"] = 3] = "ServerUnavailable";
+    ConnectReturnCode[ConnectReturnCode["BadUsernameOrPassword"] = 4] = "BadUsernameOrPassword";
+    ConnectReturnCode[ConnectReturnCode["NotAuthorized"] = 5] = "NotAuthorized";
+})(ConnectReturnCode = exports.ConnectReturnCode || (exports.ConnectReturnCode = {}));
 //# sourceMappingURL=connect.response.packet.js.map

dist/packets/disconnect.request.packet.d.ts

@@ -1,4 +0,2 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class DisconnectRequestPacket extends MqttPacket {
-    constructor();
-}
+import { PacketWriteResult } from '../mqtt.packet';
+export declare function writeDisconnectRequestPacket(): PacketWriteResult;

dist/packets/disconnect.request.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_packet_1 = require("../mqtt.packet");
-const mqtt_constants_1 = require("../mqtt.constants");
-class DisconnectRequestPacket extends mqtt_packet_1.MqttPacket {
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_DISCONNECT);
-    }
+exports.writeDisconnectRequestPacket = void 0;
+function writeDisconnectRequestPacket() {
+    return {};
 }
-exports.DisconnectRequestPacket = DisconnectRequestPacket;
+exports.writeDisconnectRequestPacket = writeDisconnectRequestPacket;
 //# sourceMappingURL=disconnect.request.packet.js.map

dist/packets/index.d.ts

@@ -15,1 +15,3 @@ export * from './connect.request.packet';
 export * from './unsubscribe.response.packet';
+export * from './packet-reader';
+export * from './packet-writer';

dist/packets/index.js

 "use strict";
-function __export(m) {
-    for (var p in m) if (!exports.hasOwnProperty(p)) exports[p] = m[p];
-}
+var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    Object.defineProperty(o, k2, { enumerable: true, get: function() { return m[k]; } });
+}) : (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    o[k2] = m[k];
+}));
+var __exportStar = (this && this.__exportStar) || function(m, exports) {
+    for (var p in m) if (p !== "default" && !exports.hasOwnProperty(p)) __createBinding(exports, m, p);
+};
 Object.defineProperty(exports, "__esModule", { value: true });
-__export(require("./connect.request.packet"));
-__export(require("./connect.response.packet"));
-__export(require("./disconnect.request.packet"));
-__export(require("./ping.request.packet"));
-__export(require("./ping.response.packet"));
-__export(require("./publish.ack.packet"));
-__export(require("./publish.complete.packet"));
-__export(require("./publish.received.packet"));
-__export(require("./publish.release.packet"));
-__export(require("./publish.request.packet"));
-__export(require("./subscribe.request.packet"));
-__export(require("./subscribe.response.packet"));
-__export(require("./unsubscribe.request.packet"));
-__export(require("./unsubscribe.response.packet"));
+__exportStar(require("./connect.request.packet"), exports);
+__exportStar(require("./connect.response.packet"), exports);
+__exportStar(require("./disconnect.request.packet"), exports);
+__exportStar(require("./ping.request.packet"), exports);
+__exportStar(require("./ping.response.packet"), exports);
+__exportStar(require("./publish.ack.packet"), exports);
+__exportStar(require("./publish.complete.packet"), exports);
+__exportStar(require("./publish.received.packet"), exports);
+__exportStar(require("./publish.release.packet"), exports);
+__exportStar(require("./publish.request.packet"), exports);
+__exportStar(require("./subscribe.request.packet"), exports);
+__exportStar(require("./subscribe.response.packet"), exports);
+__exportStar(require("./unsubscribe.request.packet"), exports);
+__exportStar(require("./unsubscribe.response.packet"), exports);
+__exportStar(require("./packet-reader"), exports);
+__exportStar(require("./packet-writer"), exports);
 //# sourceMappingURL=index.js.map

dist/packets/ping.request.packet.d.ts

@@ -1,4 +0,2 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class PingRequestPacket extends MqttPacket {
-    constructor();
-}
+import { PacketWriteResult } from '../mqtt.packet';
+export declare function writePingRequestPacket(): PacketWriteResult;

dist/packets/ping.request.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_packet_1 = require("../mqtt.packet");
-const mqtt_constants_1 = require("../mqtt.constants");
-class PingRequestPacket extends mqtt_packet_1.MqttPacket {
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_PINGREQ);
-    }
+exports.writePingRequestPacket = void 0;
+function writePingRequestPacket() {
+    return {};
 }
-exports.PingRequestPacket = PingRequestPacket;
+exports.writePingRequestPacket = writePingRequestPacket;
 //# sourceMappingURL=ping.request.packet.js.map

dist/packets/ping.response.packet.d.ts

@@ -1,4 +0,3 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class PingResponsePacket extends MqttPacket {
-    constructor();
+export declare class PingResponsePacket {
 }
+export declare function readPingResponsePacket(_: unknown, remaining: number): PingResponsePacket;

dist/packets/ping.response.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_packet_1 = require("../mqtt.packet");
-const mqtt_constants_1 = require("../mqtt.constants");
-class PingResponsePacket extends mqtt_packet_1.MqttPacket {
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_PINGRESP);
-    }
+exports.readPingResponsePacket = exports.PingResponsePacket = void 0;
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class PingResponsePacket {
 }
 exports.PingResponsePacket = PingResponsePacket;
+function readPingResponsePacket(_, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 0);
+    return new PingResponsePacket();
+}
+exports.readPingResponsePacket = readPingResponsePacket;
 //# sourceMappingURL=ping.response.packet.js.map

dist/packets/publish.ack.packet.d.ts

@@ -1,5 +0,8 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class PublishAckPacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    constructor();
+import { IdentifierPacket, PacketWriteResult } from '../mqtt.packet';
+import { PacketStream } from '../packet-stream';
+import { IdentifierData } from '../mqtt.types';
+export declare class PublishAckPacket extends IdentifierPacket {
 }
+export declare function writePublishAckPacket(stream: PacketStream, options: PublishAckPacketOptions): PacketWriteResult;
+export declare function readPublishAckPacket(stream: PacketStream, remaining: number): PublishAckPacket;
+export declare type PublishAckPacketOptions = IdentifierData;

dist/packets/publish.ack.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
+exports.readPublishAckPacket = exports.writePublishAckPacket = exports.PublishAckPacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
-class PublishAckPacket extends mqtt_packet_1.MqttPacket {
-    get hasIdentifier() {
-        return true;
-    }
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_PUBACK);
-    }
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class PublishAckPacket extends mqtt_packet_1.IdentifierPacket {
 }
 exports.PublishAckPacket = PublishAckPacket;
+function writePublishAckPacket(stream, options) {
+    stream.writeWord(options.identifier);
+    return { identifier: options.identifier };
+}
+exports.writePublishAckPacket = writePublishAckPacket;
+function readPublishAckPacket(stream, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 2);
+    return new PublishAckPacket(stream.readWord());
+}
+exports.readPublishAckPacket = readPublishAckPacket;
 //# sourceMappingURL=publish.ack.packet.js.map

dist/packets/publish.complete.packet.d.ts

@@ -1,5 +0,8 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class PublishCompletePacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    constructor();
+import { IdentifierPacket, PacketWriteResult } from '../mqtt.packet';
+import { PacketStream } from '../packet-stream';
+import { IdentifierData } from '../mqtt.types';
+export declare class PublishCompletePacket extends IdentifierPacket {
 }
+export declare function writePublishCompletePacket(stream: PacketStream, options: PublishCompletePacketOptions): PacketWriteResult;
+export declare function readPublishCompletePacket(stream: PacketStream, remaining: number): PublishCompletePacket;
+export declare type PublishCompletePacketOptions = IdentifierData;

dist/packets/publish.complete.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
+exports.readPublishCompletePacket = exports.writePublishCompletePacket = exports.PublishCompletePacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
-class PublishCompletePacket extends mqtt_packet_1.MqttPacket {
-    get hasIdentifier() {
-        return true;
-    }
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_PUBCOMP);
-    }
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class PublishCompletePacket extends mqtt_packet_1.IdentifierPacket {
 }
 exports.PublishCompletePacket = PublishCompletePacket;
+function writePublishCompletePacket(stream, options) {
+    stream.writeWord(options.identifier);
+    return { identifier: options.identifier };
+}
+exports.writePublishCompletePacket = writePublishCompletePacket;
+function readPublishCompletePacket(stream, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 2);
+    return new PublishCompletePacket(stream.readWord());
+}
+exports.readPublishCompletePacket = readPublishCompletePacket;
 //# sourceMappingURL=publish.complete.packet.js.map

dist/packets/publish.received.packet.d.ts

@@ -1,5 +0,8 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class PublishReceivedPacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    constructor();
+import { IdentifierPacket, PacketWriteResult } from '../mqtt.packet';
+import { PacketStream } from '../packet-stream';
+import { IdentifierData } from '../mqtt.types';
+export declare class PublishReceivedPacket extends IdentifierPacket {
 }
+export declare function writePublishReceivedPacket(stream: PacketStream, options: PublishReceivedPacketOptions): PacketWriteResult;
+export declare function readPublishReceivedPacket(stream: PacketStream, remaining: number): PublishReceivedPacket;
+export declare type PublishReceivedPacketOptions = IdentifierData;

dist/packets/publish.received.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
+exports.readPublishReceivedPacket = exports.writePublishReceivedPacket = exports.PublishReceivedPacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
-class PublishReceivedPacket extends mqtt_packet_1.MqttPacket {
-    get hasIdentifier() {
-        return true;
-    }
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_PUBREC);
-    }
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class PublishReceivedPacket extends mqtt_packet_1.IdentifierPacket {
 }
 exports.PublishReceivedPacket = PublishReceivedPacket;
+function writePublishReceivedPacket(stream, options) {
+    stream.writeWord(options.identifier);
+    return { identifier: options.identifier };
+}
+exports.writePublishReceivedPacket = writePublishReceivedPacket;
+function readPublishReceivedPacket(stream, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 2);
+    return new PublishReceivedPacket(stream.readWord());
+}
+exports.readPublishReceivedPacket = readPublishReceivedPacket;
 //# sourceMappingURL=publish.received.packet.js.map

dist/packets/publish.release.packet.d.ts

@@ -1,5 +0,8 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class PublishReleasePacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    constructor(identifier?: number);
+import { IdentifierPacket, PacketWriteResult } from '../mqtt.packet';
+import { PacketStream } from '../packet-stream';
+import { IdentifierData } from '../mqtt.types';
+export declare class PublishReleasePacket extends IdentifierPacket {
 }
+export declare function writePublishReleasePacket(stream: PacketStream, options: PublishReleasedPacketOptions): PacketWriteResult;
+export declare function readPublishReleasePacket(stream: PacketStream, remaining: number): PublishReleasePacket;
+export declare type PublishReleasedPacketOptions = IdentifierData;

dist/packets/publish.release.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
+exports.readPublishReleasePacket = exports.writePublishReleasePacket = exports.PublishReleasePacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
-class PublishReleasePacket extends mqtt_packet_1.MqttPacket {
-    get hasIdentifier() {
-        return true;
-    }
-    constructor(identifier) {
-        super(mqtt_constants_1.PacketTypes.TYPE_PUBREL);
-        this.packetFlags = 2;
-        this.identifier = (identifier !== null && identifier !== void 0 ? identifier : mqtt_packet_1.MqttPacket.generateIdentifier());
-    }
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class PublishReleasePacket extends mqtt_packet_1.IdentifierPacket {
 }
 exports.PublishReleasePacket = PublishReleasePacket;
+function writePublishReleasePacket(stream, options) {
+    stream.writeWord(options.identifier);
+    return { flags: 2, identifier: options.identifier };
+}
+exports.writePublishReleasePacket = writePublishReleasePacket;
+function readPublishReleasePacket(stream, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 2);
+    return new PublishReleasePacket(stream.readWord());
+}
+exports.readPublishReleasePacket = readPublishReleasePacket;
 //# sourceMappingURL=publish.release.packet.js.map

dist/packets/publish.request.packet.d.ts

 /// <reference types="node" />
 import { PacketStream } from '../packet-stream';
-import { MqttPacket } from '../mqtt.packet';
-export declare class PublishRequestPacket extends MqttPacket {
-    get payload(): Buffer;
-    get topic(): string;
-    set topic(value: string);
+import { IdentifierPacket, PacketWriteResult } from '../mqtt.packet';
+export declare class PublishRequestPacket extends IdentifierPacket {
+    flags: number;
+    topic: string;
+    payload: Buffer;
     get duplicate(): boolean;
-    set duplicate(val: boolean);
-    get retained(): boolean;
-    set retained(val: boolean);
-    get qosLevel(): number;
-    set qosLevel(val: number);
-    get hasIdentifier(): boolean;
-    protected get inlineIdentifier(): boolean;
-    private _topic;
-    private _payload;
-    constructor(topic?: string, payload?: Buffer | string | undefined, qos?: number);
-    read(stream: PacketStream): void;
-    write(stream: PacketStream): void;
+    get qos(): 0 | 1 | 2;
+    get retain(): boolean;
+    constructor(flags: number, topic: string, identifier: number | undefined, payload: Buffer);
 }
+export interface PublishPacketOptions {
+    topic: string;
+    qos?: number;
+    duplicate?: boolean;
+    retain?: boolean;
+    payload?: string | Buffer;
+    identifier?: number;
+}
+export declare function writePublishRequestPacket(stream: PacketStream, options: PublishPacketOptions): PacketWriteResult;
+export declare function readPublishRequestPacket(stream: PacketStream, remainingLength: number, flags: number): PublishRequestPacket;

dist/packets/publish.request.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
-const packet_stream_1 = require("../packet-stream");
+exports.readPublishRequestPacket = exports.writePublishRequestPacket = exports.PublishRequestPacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
 const errors_1 = require("../errors");
-class PublishRequestPacket extends mqtt_packet_1.MqttPacket {
-    constructor(topic, payload, qos) {
-        super(mqtt_constants_1.PacketTypes.TYPE_PUBLISH);
-        this._topic = (topic !== null && topic !== void 0 ? topic : '');
-        this._payload = payload ? (payload instanceof Buffer ? payload : Buffer.from(payload)) : Buffer.from([]);
-        this.qosLevel = (qos !== null && qos !== void 0 ? qos : 0);
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class PublishRequestPacket extends mqtt_packet_1.IdentifierPacket {
+    constructor(flags, topic, identifier, payload) {
+        super(identifier !== null && identifier !== void 0 ? identifier : -1);
+        this.flags = flags;
+        this.topic = topic;
+        this.payload = payload;
+        if (((flags & 0b0110) >> 1) > 2)
+            throw new Error('Invalid QoS');
     }
-    get payload() {
-        return this._payload;
-    }
-    get topic() {
-        return this._topic;
-    }
-    set topic(value) {
-        this.assertValidString(value);
-        this._topic = value;
-    }
     get duplicate() {
-        return (this.packetFlags & 8) === 8;
+        return !!(this.flags & 0b1000);
     }
-    set duplicate(val) {
-        if (val) {
-            this.packetFlags |= 8;
-        }
-        else {
-            this.packetFlags &= ~8;
-        }
+    get qos() {
+        return ((this.flags & 0b0110) >> 1);
     }
-    get retained() {
-        return (this.packetFlags & 1) === 1;
+    get retain() {
+        return !!(this.flags & 0b0001);
     }
-    set retained(val) {
-        if (val) {
-            this.packetFlags |= 1;
-        }
-        else {
-            this.packetFlags &= ~1;
-        }
+}
+exports.PublishRequestPacket = PublishRequestPacket;
+function writePublishRequestPacket(stream, options) {
+    var _a, _b;
+    options.qos = (_a = options.qos) !== null && _a !== void 0 ? _a : 0;
+    stream.writeString(options.topic);
+    if (options.qos > 2)
+        throw new Error('Unsupported QoS');
+    if (options.qos > 0) {
+        if (!options.identifier)
+            throw new Error('Expected identifier for QoS != 0');
+        stream.writeWord(options.identifier);
     }
-    get qosLevel() {
-        return (this.packetFlags & 6) >> 1;
-    }
-    set qosLevel(val) {
-        this.assertValidQosLevel(val);
-        this.packetFlags |= (val & 3) << 1;
-    }
-    get hasIdentifier() {
-        return !!this.qosLevel;
-    }
-    get inlineIdentifier() {
-        return true;
-    }
-    read(stream) {
-        super.read(stream);
-        const lastPos = stream.position;
-        this._topic = stream.readString();
-        this.readIdentifier(stream);
-        const payloadLength = this.remainingPacketLength - (stream.position - lastPos);
-        if (payloadLength === 0)
-            return;
-        if (payloadLength > stream.length - stream.position)
-            throw new errors_1.EndOfStreamError();
-        this._payload = stream.read(payloadLength);
-    }
-    write(stream) {
-        const data = this.writeIdentifier(packet_stream_1.PacketStream.empty().writeString(this._topic)).write(this._payload);
-        this.remainingPacketLength = data.length;
-        super.write(stream);
-        stream.write(data.data);
-    }
+    stream.write(mqtt_utilities_1.toBuffer((_b = options.payload) !== null && _b !== void 0 ? _b : Buffer.alloc(0)));
+    return {
+        flags: (Number(!!options.duplicate) << 3) | ((options.qos & 0x3) << 1) | Number(!!options.retain),
+        identifier: options.identifier,
+    };
 }
-exports.PublishRequestPacket = PublishRequestPacket;
+exports.writePublishRequestPacket = writePublishRequestPacket;
+function readPublishRequestPacket(stream, remainingLength, flags) {
+    const startPos = stream.position;
+    const topic = stream.readString();
+    const identifier = (flags & 0b0110) ? stream.readWord() : undefined;
+    const payloadLength = remainingLength - (stream.position - startPos);
+    if (payloadLength > stream.length - stream.position)
+        throw new errors_1.EndOfStreamError();
+    const payload = payloadLength > 0 ? stream.read(payloadLength) : Buffer.alloc(0);
+    return new PublishRequestPacket(flags, topic, identifier, payload);
+}
+exports.readPublishRequestPacket = readPublishRequestPacket;
 //# sourceMappingURL=publish.request.packet.js.map

dist/packets/subscribe.request.packet.d.ts

 import { PacketStream } from '../packet-stream';
-import { MqttPacket } from '../mqtt.packet';
-export declare class SubscribeRequestPacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    get qosLevel(): number;
-    set qosLevel(value: number);
-    get topic(): string;
-    set topic(value: string);
-    private _topic;
-    private _qosLevel;
-    constructor(topic?: string, qosLevel?: number);
-    read(): void;
-    write(stream: PacketStream): void;
+import { PacketWriteResult } from '../mqtt.packet';
+export interface SubscribePacketOptions {
+    subscriptions: Array<{
+        topic: string;
+        qos?: number;
+    }>;
+    identifier: number;
 }
+export declare function writeSubscribePacket(stream: PacketStream, options: SubscribePacketOptions): PacketWriteResult;

dist/packets/subscribe.request.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
-const packet_stream_1 = require("../packet-stream");
-const mqtt_packet_1 = require("../mqtt.packet");
-const errors_1 = require("../errors");
-class SubscribeRequestPacket extends mqtt_packet_1.MqttPacket {
-    constructor(topic, qosLevel = 1) {
-        super(mqtt_constants_1.PacketTypes.TYPE_SUBSCRIBE);
-        this.assertValidQosLevel(qosLevel);
-        this.assertValidString((topic !== null && topic !== void 0 ? topic : ''));
-        this._topic = (topic !== null && topic !== void 0 ? topic : '');
-        this._qosLevel = qosLevel;
-        this.packetFlags = 2;
+exports.writeSubscribePacket = void 0;
+function writeSubscribePacket(stream, options) {
+    var _a;
+    stream.writeWord(options.identifier);
+    if (options.subscriptions.length === 0)
+        throw new Error('The payload of a SUBSCRIBE packet MUST contain at least one Topic Filter / QoS pair');
+    for (const sub of options.subscriptions) {
+        if (sub.qos && sub.qos > 2)
+            throw new Error('invalid QoS');
+        stream.writeString(sub.topic).writeByte((_a = sub.qos) !== null && _a !== void 0 ? _a : 0);
     }
-    get hasIdentifier() {
-        return true;
-    }
-    get qosLevel() {
-        return this._qosLevel;
-    }
-    set qosLevel(value) {
-        this.assertValidQosLevel(value);
-        this._qosLevel = value;
-    }
-    get topic() {
-        return this._topic;
-    }
-    set topic(value) {
-        this.assertValidString(value);
-        this._topic = value;
-    }
-    read() {
-        throw new errors_1.InvalidDirectionError('read');
-    }
-    write(stream) {
-        const data = packet_stream_1.PacketStream.empty()
-            .writeString(this._topic)
-            .writeByte(this._qosLevel);
-        this.remainingPacketLength = data.length;
-        super.write(stream);
-        stream.write(data.data);
-    }
+    return { flags: 2, identifier: options.identifier };
 }
-exports.SubscribeRequestPacket = SubscribeRequestPacket;
+exports.writeSubscribePacket = writeSubscribePacket;
 //# sourceMappingURL=subscribe.request.packet.js.map

dist/packets/subscribe.response.packet.d.ts

 import { PacketStream } from '../packet-stream';
-import { MqttPacket } from '../mqtt.packet';
-export declare class SubscribeResponsePacket extends MqttPacket {
-    get returnCodes(): number[];
-    set returnCodes(value: number[]);
-    get hasIdentifier(): boolean;
-    private static readonly qosLevels;
-    private _returnCodes;
-    constructor();
-    read(stream: PacketStream): void;
-    write(): void;
-    isError(returnCode: number): boolean;
-    getReturnCodeName(returnCode: 0 | 1 | 2 | 128): any;
-    protected assertValidReturnCode(returnCode: number): void;
+import { IdentifierPacket } from '../mqtt.packet';
+export declare class SubscribeResponsePacket extends IdentifierPacket {
+    returnCodes: SubscribeReturnCode[];
+    get anyError(): boolean;
+    constructor(identifier: number, returnCodes: SubscribeReturnCode[]);
 }
+export declare function readSubscribeResponsePacket(stream: PacketStream, remainingLength: number): SubscribeResponsePacket;
+export declare enum SubscribeReturnCode {
+    MaxQoS0 = 0,
+    MaxQoS1 = 1,
+    MaxQoS2 = 2,
+    Fail = 128
+}

dist/packets/subscribe.response.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
+exports.SubscribeReturnCode = exports.readSubscribeResponsePacket = exports.SubscribeResponsePacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
 const errors_1 = require("../errors");
-class SubscribeResponsePacket extends mqtt_packet_1.MqttPacket {
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_SUBACK);
+class SubscribeResponsePacket extends mqtt_packet_1.IdentifierPacket {
+    constructor(identifier, returnCodes) {
+        super(identifier);
+        this.returnCodes = returnCodes;
     }
-    get returnCodes() {
-        return this._returnCodes;
+    get anyError() {
+        return !this.returnCodes.every(x => x !== SubscribeReturnCode.Fail);
     }
-    set returnCodes(value) {
-        value.forEach(e => this.assertValidReturnCode(e));
-        this._returnCodes = value;
+}
+exports.SubscribeResponsePacket = SubscribeResponsePacket;
+function readSubscribeResponsePacket(stream, remainingLength) {
+    const identifier = stream.readWord();
+    const returnCodes = Array.from(stream.read(remainingLength - 2));
+    if (!returnCodes.every(code => code === SubscribeReturnCode.MaxQoS0 ||
+        code === SubscribeReturnCode.MaxQoS1 ||
+        code === SubscribeReturnCode.MaxQoS2 ||
+        code === SubscribeReturnCode.Fail)) {
+        throw new errors_1.MalformedPacketError('Received invalid return codes');
     }
-    get hasIdentifier() {
-        return true;
-    }
-    read(stream) {
-        super.read(stream);
-        const returnCodeLen = this.remainingPacketLength - 2;
-        this._returnCodes = [];
-        for (let i = 0; i < returnCodeLen; i++) {
-            const code = stream.readByte();
-            this.assertValidReturnCode(code);
-            this._returnCodes.push(code);
-        }
-    }
-    write() {
-        throw new errors_1.InvalidDirectionError('write');
-    }
-    isError(returnCode) {
-        return returnCode === 128;
-    }
-    getReturnCodeName(returnCode) {
-        // @ts-ignore - this is valid
-        return SubscribeResponsePacket.qosLevels[`q${returnCode.toString()}`];
-    }
-    assertValidReturnCode(returnCode) {
-        if (returnCode & 124) {
-            throw new Error(`Invalid return code: ${returnCode}`);
-        }
-    }
+    return new SubscribeResponsePacket(identifier, returnCodes);
 }
-exports.SubscribeResponsePacket = SubscribeResponsePacket;
-SubscribeResponsePacket.qosLevels = {
-    q0: 'Max QoS 0',
-    q1: 'Max QoS 1',
-    q2: 'Max QoS 2',
-    q128: 'Failure',
-};
+exports.readSubscribeResponsePacket = readSubscribeResponsePacket;
+var SubscribeReturnCode;
+(function (SubscribeReturnCode) {
+    SubscribeReturnCode[SubscribeReturnCode["MaxQoS0"] = 0] = "MaxQoS0";
+    SubscribeReturnCode[SubscribeReturnCode["MaxQoS1"] = 1] = "MaxQoS1";
+    SubscribeReturnCode[SubscribeReturnCode["MaxQoS2"] = 2] = "MaxQoS2";
+    SubscribeReturnCode[SubscribeReturnCode["Fail"] = 128] = "Fail";
+})(SubscribeReturnCode = exports.SubscribeReturnCode || (exports.SubscribeReturnCode = {}));
 //# sourceMappingURL=subscribe.response.packet.js.map

dist/packets/unsubscribe.request.packet.d.ts

 import { PacketStream } from '../packet-stream';
-import { MqttPacket } from '../mqtt.packet';
-export declare class UnsubscribeRequestPacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    get topic(): string;
-    set topic(value: string);
-    private _topic;
-    constructor(topic?: string);
-    read(): void;
-    write(stream: PacketStream): void;
+import { PacketWriteResult } from '../mqtt.packet';
+export interface UnsubscribePacketOptions {
+    topics: string[] | string;
+    identifier: number;
 }
+export declare function writeUnsubscribePacket(stream: PacketStream, options: UnsubscribePacketOptions): PacketWriteResult;

dist/packets/unsubscribe.request.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
-const packet_stream_1 = require("../packet-stream");
-const mqtt_packet_1 = require("../mqtt.packet");
-const errors_1 = require("../errors");
-class UnsubscribeRequestPacket extends mqtt_packet_1.MqttPacket {
-    constructor(topic) {
-        super(mqtt_constants_1.PacketTypes.TYPE_UNSUBSCRIBE);
-        this.packetFlags = 2;
-        this.assertValidString((topic !== null && topic !== void 0 ? topic : ''));
-        this._topic = (topic !== null && topic !== void 0 ? topic : '');
-    }
-    get hasIdentifier() {
-        return true;
-    }
-    get topic() {
-        return this._topic;
-    }
-    set topic(value) {
-        this.assertValidString(value);
-        this._topic = value;
-    }
-    read() {
-        throw new errors_1.InvalidDirectionError('read');
-    }
-    write(stream) {
-        const data = packet_stream_1.PacketStream.empty().writeString(this._topic);
-        this.remainingPacketLength = data.length;
-        super.write(stream);
-        stream.write(data.data);
-    }
+exports.writeUnsubscribePacket = void 0;
+function writeUnsubscribePacket(stream, options) {
+    stream.writeWord(options.identifier);
+    const topics = typeof options.topics === 'string' ? [options.topics] : options.topics;
+    if (topics.length === 0)
+        throw new Error('The Payload of an UNSUBSCRIBE packet MUST contain at least one Topic Filter');
+    for (const topic of topics)
+        stream.writeString(topic);
+    return { flags: 2, identifier: options.identifier };
 }
-exports.UnsubscribeRequestPacket = UnsubscribeRequestPacket;
+exports.writeUnsubscribePacket = writeUnsubscribePacket;
 //# sourceMappingURL=unsubscribe.request.packet.js.map

dist/packets/unsubscribe.response.packet.d.ts

@@ -1,5 +0,5 @@
-import { MqttPacket } from '../mqtt.packet';
-export declare class UnsubscribeResponsePacket extends MqttPacket {
-    get hasIdentifier(): boolean;
-    constructor();
+import { IdentifierPacket } from '../mqtt.packet';
+import { PacketStream } from '../packet-stream';
+export declare class UnsubscribeResponsePacket extends IdentifierPacket {
 }
+export declare function readUnsubscribePacket(stream: PacketStream, remaining: number): UnsubscribeResponsePacket;

dist/packets/unsubscribe.response.packet.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
-const mqtt_constants_1 = require("../mqtt.constants");
+exports.readUnsubscribePacket = exports.UnsubscribeResponsePacket = void 0;
 const mqtt_packet_1 = require("../mqtt.packet");
-class UnsubscribeResponsePacket extends mqtt_packet_1.MqttPacket {
-    get hasIdentifier() {
-        return true;
-    }
-    constructor() {
-        super(mqtt_constants_1.PacketTypes.TYPE_UNSUBACK);
-    }
+const mqtt_utilities_1 = require("../mqtt.utilities");
+class UnsubscribeResponsePacket extends mqtt_packet_1.IdentifierPacket {
 }
 exports.UnsubscribeResponsePacket = UnsubscribeResponsePacket;
+function readUnsubscribePacket(stream, remaining) {
+    mqtt_utilities_1.expectRemainingLength(remaining, 2);
+    return new UnsubscribeResponsePacket(stream.readWord());
+}
+exports.readUnsubscribePacket = readUnsubscribePacket;
 //# sourceMappingURL=unsubscribe.response.packet.js.map

dist/transport/index.js

 "use strict";
-function __export(m) {
-    for (var p in m) if (!exports.hasOwnProperty(p)) exports[p] = m[p];
-}
+var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    Object.defineProperty(o, k2, { enumerable: true, get: function() { return m[k]; } });
+}) : (function(o, m, k, k2) {
+    if (k2 === undefined) k2 = k;
+    o[k2] = m[k];
+}));
+var __exportStar = (this && this.__exportStar) || function(m, exports) {
+    for (var p in m) if (p !== "default" && !exports.hasOwnProperty(p)) __createBinding(exports, m, p);
+};
 Object.defineProperty(exports, "__esModule", { value: true });
-__export(require("./transport"));
-__export(require("./tls.transport"));
-__export(require("./tcp.transport"));
-__export(require("./websocket.transport"));
+__exportStar(require("./transport"), exports);
+__exportStar(require("./tls.transport"), exports);
+__exportStar(require("./tcp.transport"), exports);
+__exportStar(require("./websocket.transport"), exports);
 //# sourceMappingURL=index.js.map

dist/transport/tcp.transport.d.ts

 /// <reference types="node" />
 import { Transport } from './transport';
-export declare class TcpTransport extends Transport<{
-    url: string;
-    enableTrace?: boolean;
-}> {
-    private socket;
-    send(data: Buffer): void;
-    connect(): void;
-    disconnect(): void;
+import { Socket } from 'net';
+export interface TcpTransportOptions {
+    host: string;
+    port: number;
 }
+export declare class TcpTransport extends Transport<TcpTransportOptions> {
+    duplex: Socket;
+    constructor(options: TcpTransportOptions);
+    reset(): void;
+    connect(): Promise<void>;
+}

dist/transport/tcp.transport.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.TcpTransport = void 0;
 const transport_1 = require("./transport");
-const URL = require("url");
 const net_1 = require("net");
 class TcpTransport extends transport_1.Transport {
-    send(data) {
-        this.socket.write(data);
+    constructor(options) {
+        super(options);
+        this.reset();
     }
+    reset() {
+        this.duplex = new net_1.Socket();
+        this.duplex.setNoDelay(true);
+        // buffer packets
+        this.duplex.cork();
+    }
     connect() {
-        var _a;
-        const url = URL.parse(this.options.url);
-        this.socket = net_1.connect({
-            host: (_a = url.hostname, (_a !== null && _a !== void 0 ? _a : '')),
-            port: Number(url.port),
-            timeout: 0,
-        });
-        this.socket.on('error', e => this.callbacks.error(e));
-        this.socket.on('end', () => this.callbacks.disconnect());
-        this.socket.on('close', () => this.callbacks.disconnect());
-        this.socket.on('connect', () => this.callbacks.connect());
-        this.socket.on('timeout', () => this.callbacks.disconnect());
-        this.socket.on('data', res => this.callbacks.data(res));
+        return new Promise(resolve => this.duplex.connect(this.options.port, this.options.host, () => {
+            // flush
+            this.duplex.uncork();
+            resolve();
+        }));
     }
-    disconnect() {
-        this.socket.removeAllListeners('close');
-        this.socket.end();
-    }
 }
 exports.TcpTransport = TcpTransport;
 //# sourceMappingURL=tcp.transport.js.map

dist/transport/tls.transport.d.ts

 /// <reference types="node" />
 import { Transport } from './transport';
-export declare class TlsTransport extends Transport<{
-    url: string;
-    enableTrace?: boolean;
-}> {
-    private socket;
-    send(data: Buffer): void;
-    connect(): void;
-    disconnect(): void;
+import { ConnectionOptions, TLSSocket } from 'tls';
+export interface TlsTransportOptions {
+    host: string;
+    port: number;
+    additionalOptions?: ConnectionOptions;
 }
+export declare class TlsTransport extends Transport<TlsTransportOptions> {
+    duplex: TLSSocket;
+    private underlyingSocket;
+    constructor(options: TlsTransportOptions);
+    reset(): void;
+    connect(): Promise<void>;
+}

dist/transport/tls.transport.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.TlsTransport = void 0;
 const transport_1 = require("./transport");
 const tls_1 = require("tls");
-const URL = require("url");
+const net_1 = require("net");
 class TlsTransport extends transport_1.Transport {
-    send(data) {
-        this.socket.write(data);
+    constructor(options) {
+        super(options);
+        this.reset();
     }
+    reset() {
+        this.underlyingSocket = new net_1.Socket();
+        this.underlyingSocket.setNoDelay(true);
+        this.duplex = new tls_1.TLSSocket(this.underlyingSocket);
+        this.duplex.setNoDelay(true);
+        // buffer packets until connect()
+        this.duplex.cork();
+    }
     connect() {
-        var _a;
-        const url = URL.parse(this.options.url);
-        this.socket = tls_1.connect({
-            host: (_a = url.hostname, (_a !== null && _a !== void 0 ? _a : '')),
-            port: Number(url.port),
-            enableTrace: !!this.options.enableTrace,
-            timeout: 0,
+        return new Promise(resolve => {
+            this.underlyingSocket.connect(this.options.port, this.options.host);
+            this.duplex.connect({
+                ...this.options.additionalOptions,
+                host: this.options.host,
+                port: this.options.port,
+            }, () => {
+                this.duplex.uncork();
+                resolve();
+            });
         });
-        this.socket.on('error', e => this.callbacks.error(e));
-        this.socket.on('end', () => this.callbacks.disconnect());
-        this.socket.on('close', () => this.callbacks.disconnect());
-        this.socket.on('secureConnect', () => this.callbacks.connect());
-        this.socket.on('timeout', () => this.callbacks.disconnect());
-        this.socket.on('data', res => this.callbacks.data(res));
     }
-    disconnect() {
-        this.socket.removeAllListeners('close');
-        this.socket.end();
-    }
 }
 exports.TlsTransport = TlsTransport;
 //# sourceMappingURL=tls.transport.js.map

dist/transport/transport.d.ts

 /// <reference types="node" />
+import { Duplex } from 'stream';
 export declare abstract class Transport<T> {
     protected options: T;
+    abstract duplex: Duplex;
     /**
      * This will be set by the MqttClient
      */
-    callbacks: TransportConnectOptions;
     constructor(options: T);
-    abstract connect(): void;
-    abstract disconnect(): void;
-    abstract send(data: Buffer): void;
+    abstract connect(): Promise<void>;
+    abstract reset(): void;
 }
-export interface TransportConnectOptions {
-    disconnect(data?: Error): void;
-    connect(): void;
-    data(data: Buffer): void;
-    error(e: Error): void;
-}

dist/transport/transport.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.Transport = void 0;
 class Transport {
+    /**
+     * This will be set by the MqttClient
+     */
     constructor(options) {
@@ -5,0 +9,0 @@ this.options = options;

dist/transport/websocket.transport.d.ts

 /// <reference types="node" />
 import { Transport } from './transport';
-import WebSocket = require('ws');
-export declare class WebsocketTransport extends Transport<{
+import { ClientOptions } from 'ws';
+import { Duplex } from 'stream';
+export interface WebsocketTransportOptions {
     url: string;
-    additionalOptions?: WebSocket.ClientOptions;
-}> {
-    private socket;
-    connect(): void;
-    send(data: Buffer): void;
-    disconnect(): void;
+    additionalOptions?: ClientOptions;
 }
+export declare class WebsocketTransport extends Transport<WebsocketTransportOptions> {
+    duplex: Duplex;
+    private socket?;
+    private socketStream?;
+    private readonly readable;
+    private readonly writable;
+    constructor(options: WebsocketTransportOptions);
+    reset(): void;
+    connect(): Promise<void>;
+}

dist/transport/websocket.transport.js

 "use strict";
 Object.defineProperty(exports, "__esModule", { value: true });
+exports.WebsocketTransport = void 0;
 const transport_1 = require("./transport");
 const WebSocket = require("ws");
+const stream_1 = require("stream");
+const duplexify = require("duplexify");
 class WebsocketTransport extends transport_1.Transport {
+    constructor(options) {
+        super(options);
+        this.readable = new stream_1.PassThrough();
+        this.writable = new stream_1.PassThrough();
+        this.reset();
+    }
+    reset() {
+        this.duplex = duplexify(this.writable, this.readable, { objectMode: true });
+    }
     connect() {
         this.socket = new WebSocket(this.options.url, this.options.additionalOptions);
-        this.socket.on('open', () => this.callbacks.connect());
-        this.socket.on('message', (data) => this.callbacks.data(data));
-        this.socket.on('close', () => this.callbacks.disconnect());
-        this.socket.on('error', (e) => this.callbacks.error(e));
+        this.socketStream = WebSocket.createWebSocketStream(this.socket, { objectMode: true });
+        this.socketStream.pipe(this.readable);
+        this.writable.pipe(this.socketStream);
+        return new Promise(resolve => { var _a; return (_a = this.socket) === null || _a === void 0 ? void 0 : _a.on('open', resolve); });
     }
-    send(data) {
-        this.socket.send(data);
-    }
-    disconnect() {
-        if (![this.socket.CLOSED, this.socket.CLOSING].includes(this.socket.readyState))
-            this.socket.close();
-    }
 }
 exports.WebsocketTransport = WebsocketTransport;
 //# sourceMappingURL=websocket.transport.js.map

package.json

 {
   "name": "mqtts",
-  "version": "0.1.17",
+  "version": "1.0.0-alpha.0",
   "description": "MQTT client in Typescript",
@@ -30,24 +30,30 @@ "main": "dist/index.js",
   "dependencies": {
-    "@types/ws": "^7.2.1",
+    "@types/ws": "^7.2.5",
     "debug": "^4.1.1",
-    "lodash": "^4.17.15",
-    "rxjs": "^6.5.4",
+    "duplexify": "^4.1.1",
+    "eventemitter3": "^4.0.4",
     "ts-custom-error": "^3.1.1",
     "ts-xor": "^1.0.8",
-    "ws": "^7.2.1"
+    "ws": "^7.3.0"
   },
   "devDependencies": {
+    "@types/chai": "^4.2.11",
+    "@types/chai-as-promised": "^7.1.2",
     "@types/debug": "^4.1.5",
-    "@types/jest": "^25.1.2",
-    "@types/lodash": "^4.14.149",
-    "@types/node": "^13.7.0",
-    "@typescript-eslint/eslint-plugin": "^2.19.1",
-    "@typescript-eslint/parser": "^2.19.1",
-    "eslint": "^6.8.0",
-    "eslint-config-prettier": "^6.10.0",
-    "jest": "^25.1.0",
-    "prettier": "^1.19.1",
+    "@types/duplexify": "^3.6.0",
+    "@types/jest": "^26.0.0",
+    "@types/node": "^14.0.13",
+    "@types/sinon": "^9.0.4",
+    "@typescript-eslint/eslint-plugin": "^3.3.0",
+    "@typescript-eslint/parser": "^3.3.0",
+    "chai": "^4.2.0",
+    "chai-as-promised": "^7.1.1",
+    "eslint": "^7.2.0",
+    "eslint-config-prettier": "^6.11.0",
+    "jest": "^26.0.1",
+    "prettier": "^2.0.5",
     "rimraf": "^3.0.2",
-    "ts-jest": "^25.2.0",
-    "typescript": "^3.7.5"
+    "sinon": "^9.0.2",
+    "ts-jest": "^26.1.0",
+    "typescript": "^3.9.5"
   },
@@ -54,0 +60,0 @@ "keywords": [

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

158604

increased by17.4%

Number of package files

131

increased by12.93%

Lines of code

2344

increased by11.78%

Worsened metrics

Dev dependency count

19

increased by46.15%

Number of medium supply chain risk alerts

4

increased by100%

Dependency changes

• + Addedduplexify@^4.1.1

• + Addedeventemitter3@^4.0.4

• + Addedduplexify@4.1.3(transitive)

• + Addedend-of-stream@1.4.4(transitive)

• + Addedeventemitter3@4.0.7(transitive)

• + Addedinherits@2.0.4(transitive)

• + Addedonce@1.4.0(transitive)

• + Addedreadable-stream@3.6.2(transitive)

• + Addedsafe-buffer@5.2.1(transitive)

• + Addedstream-shift@1.0.3(transitive)

• + Addedstring_decoder@1.3.0(transitive)

• + Addedutil-deprecate@1.0.2(transitive)

• + Addedwrappy@1.0.2(transitive)

• - Removedlodash@^4.17.15

• - Removedrxjs@^6.5.4

• - Removedlodash@4.17.21(transitive)

• - Removedrxjs@6.6.7(transitive)

• - Removedtslib@1.14.1(transitive)

• Updated@types/ws@^7.2.5

• Updatedws@^7.3.0