@sweepbright/iter-helpers - npm Package Compare versions

Comparing version 0.5.1 to 0.6.0

lib/Bufferize.js

@@ -30,5 +30,4 @@ "use strict";
             count = 0;
-            await outputQueue.waitDrain();
+            await outputQueue.send(result);
             cancelTimeframedFlush();
-            outputQueue.push(result);
         }
@@ -35,0 +34,0 @@ async function readInput() {

lib/Fifo.d.ts

 export interface FifoOptions {
     highWatermark?: number;
-    onSizeChange?: (size: number) => void;
 }
@@ -11,2 +10,3 @@ export declare class Fifo<T> implements AsyncIterable<T> {
      * Push an item to the queue.
+     * @deprecated Use `await send` instead
      *
@@ -41,6 +41,17 @@ * This method returns `true` if the queue is drained, i.e. the queue's
      * @returns `true` if the queue is drained, `false` otherwise
+     *
      */
     push(item: T): boolean;
+    /**
+     * @deprecated Use `await send` instead
+     */
     waitDrain(): Promise<void>;
     /**
+     * Sends an item to the fifo.
+     *
+     * Resolves as soon as the item is actually pushed.
+     * If the internal queue is full, blocks until the queue is drained.
+     */
+    send(item: T): Promise<void>;
+    /**
      * End the queue.
@@ -54,6 +65,13 @@ *
      * After the `end` method is called, calls to the `push` method will
-     * return `false` and won't result in an item being pushed.     *
+     * return `false` and won't result in an item being pushed.
+     *
+     * Resolves once all items are read by consumers.
      */
     end(): void;
-    [Symbol.asyncIterator](): AsyncGenerator<Awaited<T>, void, unknown>;
+    get stat(): {
+        data: import("@harnyk/chan").QueueStat;
+        writers: Readonly<import("@harnyk/chan").QueueStat>;
+        readers: Readonly<import("@harnyk/chan").QueueStat>;
+    };
+    [Symbol.asyncIterator](): AsyncGenerator<Awaited<T>, void, undefined>;
 }

lib/Fifo.js

 "use strict";
-var __classPrivateFieldGet = (this && this.__classPrivateFieldGet) || function (receiver, state, kind, f) {
-    if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a getter");
-    if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot read private member from an object whose class did not declare it");
-    return kind === "m" ? f : kind === "a" ? f.call(receiver) : f ? f.value : state.get(receiver);
-};
 var __classPrivateFieldSet = (this && this.__classPrivateFieldSet) || function (receiver, state, value, kind, f) {
@@ -13,19 +8,20 @@ if (kind === "m") throw new TypeError("Private method is not writable");
 };
-var _Fifo_instances, _Fifo_queue, _Fifo_ended, _Fifo_onDrain, _Fifo_onDrainPromise, _Fifo_isDrain, _Fifo_onEnd, _Fifo_waitEnd, _Fifo_onPush, _Fifo_waitPush, _Fifo_flush, _Fifo_reportSize;
+var __classPrivateFieldGet = (this && this.__classPrivateFieldGet) || function (receiver, state, kind, f) {
+    if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a getter");
+    if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot read private member from an object whose class did not declare it");
+    return kind === "m" ? f : kind === "a" ? f.call(receiver) : f ? f.value : state.get(receiver);
+};
+var _Fifo_ch;
 Object.defineProperty(exports, "__esModule", { value: true });
 exports.Fifo = void 0;
-const sleep_1 = require("./tests/sleep");
+const chan_1 = require("@harnyk/chan");
 class Fifo {
     constructor(options) {
-        _Fifo_instances.add(this);
         this.options = options;
-        _Fifo_queue.set(this, []);
-        _Fifo_ended.set(this, false);
-        _Fifo_onDrain.set(this, void 0);
-        _Fifo_onDrainPromise.set(this, void 0);
-        _Fifo_onEnd.set(this, void 0);
-        _Fifo_onPush.set(this, void 0);
+        _Fifo_ch.set(this, void 0);
+        __classPrivateFieldSet(this, _Fifo_ch, new chan_1.CompatChan(this.options?.highWatermark ?? Infinity), "f");
     }
     /**
      * Push an item to the queue.
+     * @deprecated Use `await send` instead
      *
@@ -60,30 +56,23 @@ * This method returns `true` if the queue is drained, i.e. the queue's
      * @returns `true` if the queue is drained, `false` otherwise
+     *
      */
     push(item) {
-        if (__classPrivateFieldGet(this, _Fifo_ended, "f")) {
-            return false;
-        }
-        __classPrivateFieldGet(this, _Fifo_queue, "f").push(item);
-        __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_reportSize).call(this);
-        const isDrain = __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_isDrain).call(this);
-        __classPrivateFieldGet(this, _Fifo_onPush, "f")?.call(this);
-        return isDrain;
+        return __classPrivateFieldGet(this, _Fifo_ch, "f").sendSync(item);
     }
+    /**
+     * @deprecated Use `await send` instead
+     */
     waitDrain() {
-        if (__classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_isDrain).call(this)) {
-            return (0, sleep_1.sleep)(0);
-        }
-        const onDrainPromise = __classPrivateFieldGet(this, _Fifo_onDrainPromise, "f") ??
-            new Promise((resolve) => {
-                __classPrivateFieldSet(this, _Fifo_onDrain, resolve, "f");
-            }).then(() => {
-                __classPrivateFieldSet(this, _Fifo_onDrain, undefined, "f");
-                __classPrivateFieldSet(this, _Fifo_onDrainPromise, undefined, "f");
-            });
-        if (!__classPrivateFieldGet(this, _Fifo_onDrainPromise, "f")) {
-            __classPrivateFieldSet(this, _Fifo_onDrainPromise, onDrainPromise, "f");
-        }
-        return __classPrivateFieldGet(this, _Fifo_onDrainPromise, "f");
+        return __classPrivateFieldGet(this, _Fifo_ch, "f").readySend();
     }
     /**
+     * Sends an item to the fifo.
+     *
+     * Resolves as soon as the item is actually pushed.
+     * If the internal queue is full, blocks until the queue is drained.
+     */
+    send(item) {
+        return __classPrivateFieldGet(this, _Fifo_ch, "f").send(item);
+    }
+    /**
      * End the queue.
@@ -97,46 +86,17 @@ *
      * After the `end` method is called, calls to the `push` method will
-     * return `false` and won't result in an item being pushed.     *
+     * return `false` and won't result in an item being pushed.
+     *
+     * Resolves once all items are read by consumers.
      */
     end() {
-        if (__classPrivateFieldGet(this, _Fifo_ended, "f")) {
-            return;
-        }
-        __classPrivateFieldSet(this, _Fifo_ended, true, "f");
-        const onEnd = __classPrivateFieldGet(this, _Fifo_onEnd, "f");
-        onEnd?.();
+        return __classPrivateFieldGet(this, _Fifo_ch, "f").close();
     }
-    async *[(_Fifo_queue = new WeakMap(), _Fifo_ended = new WeakMap(), _Fifo_onDrain = new WeakMap(), _Fifo_onDrainPromise = new WeakMap(), _Fifo_onEnd = new WeakMap(), _Fifo_onPush = new WeakMap(), _Fifo_instances = new WeakSet(), _Fifo_isDrain = function _Fifo_isDrain() {
-        return this.options?.highWatermark
-            ? __classPrivateFieldGet(this, _Fifo_queue, "f").length < this.options?.highWatermark
-            : true;
-    }, _Fifo_waitEnd = function _Fifo_waitEnd() {
-        if (__classPrivateFieldGet(this, _Fifo_ended, "f")) {
-            return Promise.resolve();
-        }
-        return new Promise((resolve) => (__classPrivateFieldSet(this, _Fifo_onEnd, resolve, "f")));
-    }, _Fifo_waitPush = function _Fifo_waitPush() {
-        return new Promise((resolve) => (__classPrivateFieldSet(this, _Fifo_onPush, resolve, "f")));
-    }, _Fifo_flush = async function* _Fifo_flush() {
-        while (__classPrivateFieldGet(this, _Fifo_queue, "f").length) {
-            const prevIsDrain = __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_isDrain).call(this);
-            yield __classPrivateFieldGet(this, _Fifo_queue, "f").shift();
-            __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_reportSize).call(this);
-            if (__classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_isDrain).call(this) && !prevIsDrain) {
-                __classPrivateFieldGet(this, _Fifo_onDrain, "f")?.call(this);
-            }
-        }
-    }, _Fifo_reportSize = function _Fifo_reportSize() {
-        this.options?.onSizeChange?.(__classPrivateFieldGet(this, _Fifo_queue, "f").length);
-    }, Symbol.asyncIterator)]() {
-        for (;;) {
-            yield* __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_flush).call(this);
-            await Promise.race([__classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_waitPush).call(this), __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_waitEnd).call(this)]);
-            if (__classPrivateFieldGet(this, _Fifo_ended, "f")) {
-                break;
-            }
-        }
-        yield* __classPrivateFieldGet(this, _Fifo_instances, "m", _Fifo_flush).call(this);
+    get stat() {
+        return __classPrivateFieldGet(this, _Fifo_ch, "f").stat;
     }
+    async *[(_Fifo_ch = new WeakMap(), Symbol.asyncIterator)]() {
+        yield* __classPrivateFieldGet(this, _Fifo_ch, "f");
+    }
 }
 exports.Fifo = Fifo;
 //# sourceMappingURL=Fifo.js.map

lib/Mux.js

@@ -15,4 +15,3 @@ "use strict";
             .tap(async (value) => {
-            await fifo.waitDrain();
-            fifo.push(value);
+            await fifo.send(value);
         })
@@ -19,0 +18,0 @@ .consume());

lib/tests/kitchen-sink.spec.js

@@ -32,4 +32,3 @@ "use strict";
         async function enqueueRetry(input) {
-            await queue.waitDrain();
-            return queue.push(input);
+            await queue.send(input);
         }
@@ -52,4 +51,3 @@ // Check if the result is an error
             .tap(async (batch) => {
-            await queue.waitDrain();
-            queue.push(batch);
+            await queue.send(batch);
         })
@@ -56,0 +54,0 @@ .consume();

package.json

 {
   "name": "@sweepbright/iter-helpers",
-  "version": "0.5.1",
+  "version": "0.6.0",
   "description": "",
@@ -35,3 +35,5 @@ "repository": {
   },
-  "dependencies": {},
+  "dependencies": {
+    "@harnyk/chan": "^0.1.1"
+  },
   "devDependencies": {
@@ -38,0 +40,0 @@ "@types/jest": "^28.1.6",

No alert changes

Improved metrics

Total package byte prevSize

123789

increased by9.76%

Number of package files

113

increased by2.73%

Lines of code

1862

increased by12.03%

Worsened metrics

Dependency count

1

increased byInfinity%

Dependency changes

• + Added@harnyk/chan@^0.1.1

• + Added@harnyk/chan@0.1.1(transitive)