@basic-streams/from-iterable - npm Package Compare versions

Comparing version 0.0.1 to 0.0.2

index.js

 "use strict";
-var __values = (this && this.__values) || function (o) {
-    var m = typeof Symbol === "function" && o[Symbol.iterator], i = 0;
-    if (m) return m.call(o);
-    return {
-        next: function () {
-            if (o && i >= o.length) o = void 0;
-            return { value: o && o[i++], done: !o };
-        }
-    };
+var __importDefault = (this && this.__importDefault) || function (mod) {
+    return (mod && mod.__esModule) ? mod : { "default": mod };
 };
 Object.defineProperty(exports, "__esModule", { value: true });
+var later_1 = __importDefault(require("@basic-streams/later"));
 function noop() { }
-function fromIterable(xs) {
+function getIterator(iterable) {
+    var maybeIteratorGetter = typeof Symbol === "function" && iterable[Symbol.iterator];
+    if (maybeIteratorGetter) {
+        return maybeIteratorGetter.call(iterable);
+    }
+    if (Array.isArray(iterable)) {
+        var i_1 = 0;
+        return {
+            next: function () {
+                return i_1 >= iterable.length
+                    ? { value: undefined, done: true }
+                    : { value: iterable[i_1++], done: false };
+            },
+        };
+    }
+    throw new TypeError("a value provided to fromIterable() isn't an Iterable");
+}
+function fromIterable(xs, interval, scheduler) {
+    if (scheduler === void 0) { scheduler = later_1.default; }
     return function (cb) {
-        var e_1, _a;
-        try {
-            for (var xs_1 = __values(xs), xs_1_1 = xs_1.next(); !xs_1_1.done; xs_1_1 = xs_1.next()) {
-                var x = xs_1_1.value;
-                cb(x);
+        // without interval
+        var iterator = getIterator(xs);
+        if (interval === undefined) {
+            var next_1 = iterator.next();
+            while (!next_1.done) {
+                cb(next_1.value);
+                next_1 = iterator.next();
             }
+            return noop;
         }
-        catch (e_1_1) { e_1 = { error: e_1_1 }; }
-        finally {
-            try {
-                if (xs_1_1 && !xs_1_1.done && (_a = xs_1.return)) _a.call(xs_1);
+        // with interval
+        var schedulerStream = scheduler(interval);
+        var inLoop = false;
+        var next = null;
+        var schedulerDisposer = null;
+        var firstDisposer = schedulerStream(function step() {
+            next = iterator.next();
+            if (inLoop) {
+                return;
             }
-            finally { if (e_1) throw e_1.error; }
+            inLoop = true;
+            while (next !== null) {
+                if (next.done) {
+                    return;
+                }
+                var value = next.value;
+                next = null;
+                schedulerDisposer = schedulerStream(step);
+                cb(value);
+            }
+            inLoop = false;
+        });
+        if (schedulerDisposer === null) {
+            schedulerDisposer = firstDisposer;
         }
-        return noop;
+        return function () {
+            if (schedulerDisposer !== null)
+                schedulerDisposer();
+        };
     };
 }
 exports.default = fromIterable;

index.ts

 import {Stream} from "@basic-streams/stream"
+import later from "@basic-streams/later"
 
 function noop() {}
 
-export default function fromIterable<T>(xs: Iterable<T>): Stream<T> {
+type IteratorResult<T> =
+  | {done: true; value: undefined}
+  | {done: false; value: T}
+
+interface IteratorLike<T> {
+  next(): IteratorResult<T>
+}
+
+function getIterator<T>(iterable: Iterable<T>): IteratorLike<T> {
+  const maybeIteratorGetter =
+    typeof Symbol === "function" && iterable[Symbol.iterator]
+  if (maybeIteratorGetter) {
+    return maybeIteratorGetter.call(iterable)
+  }
+  if (Array.isArray(iterable)) {
+    let i = 0
+    return {
+      next() {
+        return i >= iterable.length
+          ? {value: undefined, done: true}
+          : {value: iterable[i++], done: false}
+      },
+    }
+  }
+  throw new TypeError("a value provided to fromIterable() isn't an Iterable")
+}
+
+export default function fromIterable<T>(
+  xs: Iterable<T>,
+  interval?: number,
+  scheduler = later,
+): Stream<T> {
   return cb => {
-    for (const x of xs) {
-      cb(x)
+    // without interval
+    const iterator = getIterator(xs)
+    if (interval === undefined) {
+      let next = iterator.next()
+      while (!next.done) {
+        cb(next.value)
+        next = iterator.next()
+      }
+      return noop
     }
-    return noop
+
+    // with interval
+    const schedulerStream = scheduler(interval)
+    let inLoop = false
+    let next: null | IteratorResult<T> = null
+    let schedulerDisposer: null | (() => void) = null
+    let firstDisposer = schedulerStream(function step() {
+      next = iterator.next()
+      if (inLoop) {
+        return
+      }
+      inLoop = true
+      while (next !== null) {
+        if (next.done) {
+          return
+        }
+        const {value} = next
+        next = null
+        schedulerDisposer = schedulerStream(step)
+        cb(value)
+      }
+      inLoop = false
+    })
+    if (schedulerDisposer === null) {
+      schedulerDisposer = firstDisposer
+    }
+    return () => {
+      if (schedulerDisposer !== null) schedulerDisposer()
+    }
   }
 }

package.json

 {
   "name": "@basic-streams/from-iterable",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "description": "fromIterable operator for basic-streams",
@@ -5,0 +5,0 @@ "keywords": [

README.md

@@ -1,3 +0,9 @@
-# @basic-streams/from-iterable
+# [@basic-streams](https://github.com/rpominov/basic-streams)/from-iterable
 
 TODO: fill up README
+
+<!-- api-doc-start -->
+
+### `fromIterable<T>(xs: Iterable<T>, interval?: number, scheduler?: typeof later): Stream<T>`
+
+<!-- api-doc-end -->

test.ts

@@ -1,3 +0,4 @@
-import {EventsList, emulate, t, v} from "@basic-streams/emulation"
+import {EventsList, emulate, t, v, laterMock} from "../emulation"
 import fromIterable from "./index"
+import take from "../take"
 
@@ -8,3 +9,7 @@ expect.addSnapshotSerializer(EventsList.jestSerializer)
   const cb = jest.fn()
-  fromIterable([1, 2, 3])(cb)
+  const array = [1, 2, 3]
+  if (typeof Symbol === "function" && array[Symbol.iterator]) {
+    array[Symbol.iterator] = undefined
+  }
+  fromIterable(array)(cb)
   expect(cb.mock.calls).toMatchSnapshot()
@@ -24,1 +29,76 @@ })
 })
+
+test("when interval provided spreads values in time", () => {
+  const result = emulate(create => {
+    return fromIterable([1, 2, 3], 10, laterMock(create))
+  })
+  expect(result).toMatchSnapshot()
+})
+
+test("when interval provided disposer works", () => {
+  function unsafeTakeOne(stream) {
+    return cb => {
+      let disposer = stream(x => {
+        cb(x)
+        disposer()
+      })
+      return disposer
+    }
+  }
+  const result = emulate(create => {
+    return unsafeTakeOne(fromIterable([1, 2, 3], 10, laterMock(create)))
+  })
+  expect(result).toMatchSnapshot()
+})
+
+test("doesn't blow up call stack when scheduler is synchronous", () => {
+  function scheduler(time, value) {
+    return cb => {
+      cb(value)
+      return () => {}
+    }
+  }
+  function* generator() {
+    for (let i = 0; i <= 1000000; i++) {
+      yield i
+    }
+  }
+  const stream = fromIterable(generator(), 10, scheduler)
+  let count = 0
+  let latestValue = null
+  function cb(value) {
+    count++
+    latestValue = value
+  }
+  expect(() => stream(cb)).not.toThrow()
+  expect({count, latestValue}).toMatchSnapshot()
+})
+
+test("correctly handles disposers when scheduler runs synchronously only first time", () => {
+  const disposer = jest.fn()
+  let runned = false
+  function scheduler(time, value) {
+    return cb => {
+      if (!runned) {
+        runned = true
+        cb(value)
+        return () => {}
+      }
+      return disposer
+    }
+  }
+  fromIterable([1, 2], 0, scheduler)(() => {})()
+  expect(disposer.mock.calls).toMatchSnapshot()
+})
+
+test("doesn't drain iterable eagerly", () => {
+  function* generator() {
+    yield 1
+    throw new Error("should not be reached")
+  }
+  expect(() =>
+    emulate(create => {
+      return take(1, fromIterable(generator(), 10, laterMock(create)))
+    }),
+  ).not.toThrow()
+})

New alerts

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

Improved metrics

Total package byte prevSize

9316

increased by148.36%

Lines of code

236

increased by280.65%

Number of lines in readme file

10

increased by150%

Worsened metrics

Number of medium supply chain risk alerts

1

increased byInfinity%

No dependency changes