mqtts - npm Package Compare versions

Comparing version 1.4.0-beta.0 to 1.4.0-beta.1

dist/errors/index.d.ts

@@ -8,1 +8,2 @@ export * from './end-of-stream.error';
 export * from './illegal-state.error';
+export * from './abort.error';

dist/errors/index.js

@@ -20,2 +20,3 @@ "use strict";
 __exportStar(require("./illegal-state.error"), exports);
+__exportStar(require("./abort.error"), exports);
 //# sourceMappingURL=index.js.map

dist/mqtt.base-client.d.ts

@@ -52,13 +52,2 @@ import { RegisterClientOptions, Resolvable } from './mqtt.types';
     resolveConnectOptions(): Promise<RegisterClientOptions>;
-    private _connectPromise?;
-    private _connectResolve?;
-    private _connectReject?;
-    createConnectPromise(): Promise<void>;
-    resolveConnectPromise(): void;
-    rejectConnectPromise(e: Error): void;
-    /**
-     * Only rejects the promise if it's still pending
-     * @param {Error} e
-     */
-    rejectConnectPromiseIfPending(e: Error): void;
 }

dist/mqtt.base-client.js

@@ -6,3 +6,2 @@ "use strict";
 const EventEmitter = require("eventemitter3");
-const errors_1 = require("./errors");
 var StateId;
@@ -112,40 +111,4 @@ (function (StateId) {
     }
-    createConnectPromise() {
-        this.expectConnecting();
-        if (this._connectPromise) {
-            throw new Error('Already created a promise.');
-        }
-        this._connectPromise = new Promise((resolve, reject) => {
-            this._connectResolve = resolve;
-            this._connectReject = reject;
-        });
-        return this._connectPromise;
-    }
-    resolveConnectPromise() {
-        if (!this._connectResolve)
-            throw new errors_1.IllegalStateError('No resolve-function found');
-        this._connectResolve();
-        this._connectPromise = undefined;
-        this._connectResolve = undefined;
-        this._connectReject = undefined;
-    }
-    rejectConnectPromise(e) {
-        if (!this._connectReject)
-            throw new errors_1.IllegalStateError(`No reject-function found - Error: ${e.message}`);
-        this._connectReject(e);
-        this._connectPromise = undefined;
-        this._connectResolve = undefined;
-        this._connectReject = undefined;
-    }
-    /**
-     * Only rejects the promise if it's still pending
-     * @param {Error} e
-     */
-    rejectConnectPromiseIfPending(e) {
-        if (!this._connectReject)
-            return;
-        return this.rejectConnectPromise(e);
-    }
 }
 exports.MqttBaseClient = MqttBaseClient;
 //# sourceMappingURL=mqtt.base-client.js.map

dist/mqtt.client.d.ts

@@ -32,3 +32,2 @@ /// <reference types="node" />
     protected writer: PacketWriter<WriteMap>;
-    protected connectTimer?: TimerRef;
     protected keepAliveTimer?: TimerRef;
@@ -41,2 +40,3 @@ protected autoReconnect?: boolean | MqttAutoReconnectOptions;
     connect(options?: Resolvable<RegisterClientOptions>): Promise<any>;
+    protected createPipeline(): void;
     publish(message: MqttMessageOutgoing): Promise<MqttMessageOutgoing>;
@@ -61,3 +61,5 @@ subscribe(subscription: MqttSubscription): Promise<SubscribeReturnCode>;
     protected clearFinishedFlows(): void;
-    protected registerClient(options: RegisterClientOptions, noNewPromise?: boolean, lastFlow?: PacketFlowFunc<ReadMap, WriteMap, unknown>): Promise<any>;
+    protected stopExecutingFlows(error: Error): void;
+    protected getFlowById<T = any>(id: number | bigint): PacketFlowData<T> | undefined;
+    protected registerClient(options: RegisterClientOptions): Promise<any>;
     protected getConnectFlow(options: ConnectRequestOptions): PacketFlowFunc<ReadMap, WriteMap, unknown>;
@@ -64,0 +66,0 @@ protected updateKeepAlive(value: number): void;

dist/mqtt.client.js

@@ -75,22 +75,22 @@ "use strict";
         await this.transport.connect();
+        this.createPipeline();
+        return this.registerClient(await this.resolveConnectOptions());
+    }
+    createPipeline() {
         if (!this.transport.duplex)
             throw new errors_1.IllegalStateError('Expected transport to expose a Duplex.');
-        this.pipeline = stream_1.pipeline(this.transport.duplex, this.transformer, new stream_1.Writable({
-            write: (chunk, encoding, callback) => {
+        this.pipeline = stream_1.pipeline(this.transport.duplex, this.transformer, (async (source) => {
+            for await (const chunk of source) {
                 if (!chunk.type) {
-                    callback(new Error('Chunk is not a MqttPacket'));
-                    return;
+                    throw new Error('Chunk is not a MqttPacket');
                 }
-                this.handlePacket(chunk)
-                    .then(() => callback())
-                    .catch(callback);
-            },
-            objectMode: true,
-        }), err => {
+                await this.handlePacket(chunk);
+            }
+            return 'Source drained';
+        }) /* bad definitions */, err => {
             if (err)
                 this.emitError(err);
             if (!this.disconnected)
-                this.setDisconnected('Pipeline finished');
+                this.setDisconnected('Pipeline finished').catch(e => this.emitWarning(e));
         });
-        return this.registerClient(await this.resolveConnectOptions());
     }
@@ -160,3 +160,3 @@ publish(message) {
     stopFlow(flowId, rejection) {
-        const flow = this.activeFlows.find(f => f.flowId);
+        const flow = this.getFlowById(flowId);
         if (!flow)
@@ -192,31 +192,34 @@ return false;
     }
-    registerClient(options, noNewPromise = false, lastFlow) {
-        var _a;
-        let promise;
-        if (noNewPromise) {
-            const flow = this.activeFlows.find(x => x.flowFunc === lastFlow);
-            if (!flow) {
-                promise = Promise.reject(new Error('Could not find flow'));
-            }
-            else {
+    stopExecutingFlows(error) {
+        for (const flow of this.activeFlows) {
+            flow.resolvers.reject(error);
+            flow.finished = true;
+        }
+        this.activeFlows = [];
+    }
+    getFlowById(id) {
+        return this.activeFlows.find(f => f.flowId === id);
+    }
+    registerClient(options) {
+        var _a, _b;
+        const flow = this.getConnectFlow(options);
+        const connectPromiseFlow = this.startFlow(flow);
+        if (typeof options.connectDelay !== 'undefined') {
+            const timerId = this.executeDelayed((_a = options.connectDelay) !== null && _a !== void 0 ? _a : 2000, () => {
+                const flow = this.getFlowById(connectPromiseFlow.flowId);
+                if (!flow) {
+                    // there's no flow anymore
+                    this.stopExecuting(timerId);
+                    return;
+                }
                 const packet = flow.callbacks.start();
                 if (packet)
                     this.sendData(this.writer.write(packet.type, packet.options));
-                promise = Promise.resolve();
-            }
+            });
+            connectPromiseFlow.finally(() => this.stopExecuting(timerId))
+                // not sure why this is necessary, but it's there so no unhandledRejection is thrown
+                .catch(() => undefined);
         }
-        else {
-            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 =
-            typeof options.connectDelay === 'undefined'
-                ? undefined
-                : 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;
+        (_b = options.signal) === null || _b === void 0 ? void 0 : _b.addEventListener('abort', () => this.stopFlow(connectPromiseFlow.flowId, new errors_1.AbortError('Connecting aborted')));
+        return connectPromiseFlow;
     }
@@ -311,11 +314,7 @@ getConnectFlow(options) {
         super.reset();
-        if (this.connecting)
-            this.rejectConnectPromiseIfPending(new Error('Disconnected'));
-        if (this.connectTimer)
-            clearTimeout(this.connectTimer);
-        this.connectTimer = undefined;
-        if (this.keepAliveTimer)
+        this.stopExecutingFlows(new errors_1.AbortError('Resetting'));
+        if (this.keepAliveTimer) {
             clearInterval(this.keepAliveTimer);
-        this.keepAliveTimer = undefined;
-        this.activeFlows = [];
+            this.keepAliveTimer = undefined;
+        }
         this.transformer.reset();
@@ -326,4 +325,2 @@ }
         this.mqttDebug('Ready!');
-        if (this.connectTimer)
-            this.stopExecuting(this.connectTimer);
     }
@@ -350,4 +347,3 @@ shouldReconnect() {
         const willReconnect = this.autoReconnect && this.shouldReconnect();
-        if (this.connecting)
-            this.rejectConnectPromiseIfPending(new Error('Disconnected'));
+        this.stopExecutingFlows(new errors_1.AbortError('Client disconnected.'));
         super.setDisconnected();
@@ -354,0 +350,0 @@ this.emitDisconnect(`reason: ${reason} willReconnect: ${willReconnect}`);

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

@@ -20,4 +20,5 @@ /// <reference types="node" />
     connectDelay?: number;
+    signal?: AbortSignal;
 }
 export declare function writeConnectPacket(stream: PacketStream, options: RequiredConnectRequestOptions): PacketWriteResult;
 export declare function makeFlags(options: ConnectRequestOptions): number;

package.json

 {
   "name": "mqtts",
-  "version": "1.4.0-beta.0",
+  "version": "1.4.0-beta.1",
   "description": "MQTT client in Typescript",
@@ -11,3 +11,3 @@ "main": "dist/index.js",
   "engines": {
-    "node": ">=10.0.0"
+    "node": ">=14.0.0"
   },
@@ -45,3 +45,3 @@ "repository": {
     "@types/jest": "^26.0.0",
-    "@types/node": "^14.0.13",
+    "@types/node": "^14.14.35",
     "@types/sinon": "^9.0.4",
@@ -48,0 +48,0 @@ "@typescript-eslint/eslint-plugin": "4.18.0",

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

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

Number of package files

143

increased by2.14%

Worsened metrics

Total package byte prevSize

168359

decreased by-1.25%

Lines of code

2513

decreased by-1.45%

No dependency changes