Repository: marpple/FxTS
Pull request #17 (2021-11-23 02:59:50)

fix: concurrent handling takeUntil/takeWhile
If the requested order is different, the response order must be controlled



Patch:
diff --git a/src/Lazy/takeUntil.ts b/src/Lazy/takeUntil.ts
index 99c994da..678c07ce 100644
--- a/src/Lazy/takeUntil.ts
+++ b/src/Lazy/takeUntil.ts
@@ -2,6 +2,7 @@ import { isAsyncIterable, isIterable } from "../_internal/utils";
 import IterableInfer from "../types/IterableInfer";
 import ReturnIterableIteratorType from "../types/ReturnIterableIteratorType";
 import { AsyncFunctionException } from "../_internal/error";
+import concurrent, { isConcurrent } from "./concurrent";
 
 function* sync<A, B>(f: (a: A) => B, iterable: Iterable<A>) {
   for (const item of iterable) {
@@ -18,7 +19,7 @@ function* sync<A, B>(f: (a: A) => B, iterable: Iterable<A>) {
   }
 }
 
-function async<A, B>(
+function asyncSequential<A, B>(
   f: (a: A) => B,
   iterable: AsyncIterable<A>,
 ): AsyncIterableIterator<A> {
@@ -29,18 +30,12 @@ function async<A, B>(
       return this;
     },
     async next(_concurrent) {
-      if (end) {
-        return { done: true, value: undefined };
-      }
-
       const { done, value } = await iterator.next(_concurrent);
-
       if (done || end) {
         return { done: true, value: undefined };
       }
 
       const cond = await f(value);
-
       if (end) {
         return { done: true, value: undefined };
       }
@@ -54,6 +49,26 @@ function async<A, B>(
   };
 }
 
+function async<A, B>(
+  f: (a: A) => B,
+  iterable: AsyncIterable<A>,
+): AsyncIterableIterator<A> {
+  let _iterator: AsyncIterator<A>;
+  return {
+    async next(_concurrent: any) {
+      if (_iterator === undefined) {
+        _iterator = isConcurrent(_concurrent)
+          ? asyncSequential(f, concurrent(_concurrent.length, iterable))
+          : asyncSequential(f, iterable);
+      }
+      return _iterator.next(_concurrent);
+    },
+    [Symbol.asyncIterator]() {
+      return this;
+    },
+  };
+}
+
 /**
  * Returns Iterable/AsyncIterable that taken values until truthy when given `f` is applied.
  *
diff --git a/src/Lazy/takeWhile.ts b/src/Lazy/takeWhile.ts
index 89236d5b..5cc9a05e 100644
--- a/src/Lazy/takeWhile.ts
+++ b/src/Lazy/takeWhile.ts
@@ -2,6 +2,7 @@ import { isAsyncIterable, isIterable } from "../_internal/utils";
 import IterableInfer from "../types/IterableInfer";
 import ReturnIterableIteratorType from "../types/ReturnIterableIteratorType";
 import { AsyncFunctionException } from "../_internal/error";
+import concurrent, { isConcurrent } from "./concurrent";
 
 function* sync<A, B>(f: (a: A) => B, iterable: Iterable<A>) {
   for (const item of iterable) {
@@ -17,7 +18,7 @@ function* sync<A, B>(f: (a: A) => B, iterable: Iterable<A>) {
   }
 }
 
-function async<A, B>(
+function asyncSequential<A, B>(
   f: (a: A) => B,
   iterable: AsyncIterable<A>,
 ): AsyncIterableIterator<A> {
@@ -29,7 +30,6 @@ function async<A, B>(
     },
     async next(_concurrent) {
       const { done, value } = await iterator.next(_concurrent);
-
       if (done || end) {
         return { done: true, value: undefined };
       }
@@ -44,6 +44,26 @@ function async<A, B>(
   };
 }
 
+function async<A, B>(
+  f: (a: A) => B,
+  iterable: AsyncIterable<A>,
+): AsyncIterableIterator<A> {
+  let _iterator: AsyncIterator<A>;
+  return {
+    async next(_concurrent: any) {
+      if (_iterator === undefined) {
+        _iterator = isConcurrent(_concurrent)
+          ? asyncSequential(f, concurrent(_concurrent.length, iterable))
+          : asyncSequential(f, iterable);
+      }
+      return _iterator.next(_concurrent);
+    },
+    [Symbol.asyncIterator]() {
+      return this;
+    },
+  };
+}
+
 /**
  * Returns Iterable/AsyncIterable that taken values as long as each value satisfies the give `f`.
  *


Test patch:
diff --git a/test/Lazy/takeUntil.spec.ts b/test/Lazy/takeUntil.spec.ts
index fb46e71c..08c9dc36 100644
--- a/test/Lazy/takeUntil.spec.ts
+++ b/test/Lazy/takeUntil.spec.ts
@@ -145,5 +145,46 @@ describe("takeUntil", function () {
       await iter.next(concurrent);
       expect((mock as any).getConcurrent()).toEqual(concurrent);
     });
+
+    it("should be able to handle an error when asynchronous", async function () {
+      await expect(
+        pipe(
+          range(Infinity),
+          toAsync,
+          map((a) => delay(100, a + 10)),
+          filter((a) => a % 2 === 0),
+          takeUntil((a) => {
+            if (a > 15) {
+              throw new Error("err");
+            }
+            return false;
+          }),
+          concurrent(5),
+          toArray,
+        ),
+      ).rejects.toThrow("err");
+    }, 350);
+
+    it("should be controlled the order when concurrency", async function () {
+      const res = await pipe(
+        toAsync(
+          (function* () {
+            yield delay(500, 1);
+            yield delay(400, 2);
+            yield delay(300, 3);
+            yield delay(200, 4);
+            yield delay(100, 5);
+            yield delay(500, 6);
+            yield delay(400, 7);
+            yield delay(300, 8);
+            yield delay(200, 9);
+          })(),
+        ),
+        takeUntil((a) => a > 8),
+        concurrent(5),
+        toArray,
+      );
+      expect(res).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9]);
+    }, 1050);
   });
 });
diff --git a/test/Lazy/takeWhile.spec.ts b/test/Lazy/takeWhile.spec.ts
index 80fffeaa..c3a3735c 100644
--- a/test/Lazy/takeWhile.spec.ts
+++ b/test/Lazy/takeWhile.spec.ts
@@ -138,4 +138,45 @@ describe("takeWhile", function () {
     await iter.next(concurrent);
     expect((mock as any).getConcurrent()).toEqual(concurrent);
   });
+
+  it("should be able to handle an error when asynchronous", async function () {
+    await expect(
+      pipe(
+        range(Infinity),
+        toAsync,
+        map((a) => delay(100, a + 10)),
+        filter((a) => a % 2 === 0),
+        takeWhile((a) => {
+          if (a > 15) {
+            throw new Error("err");
+          }
+          return true;
+        }),
+        concurrent(5),
+        toArray,
+      ),
+    ).rejects.toThrow("err");
+  }, 350);
+
+  it("should be controlled the order when concurrency", async function () {
+    const res = await pipe(
+      toAsync(
+        (function* () {
+          yield delay(500, 1);
+          yield delay(400, 2);
+          yield delay(300, 3);
+          yield delay(200, 4);
+          yield delay(100, 5);
+          yield delay(500, 6);
+          yield delay(400, 7);
+          yield delay(300, 8);
+          yield delay(200, 9);
+        })(),
+      ),
+      takeWhile((a) => a < 8),
+      concurrent(5),
+      toArray,
+    );
+    expect(res).toEqual([1, 2, 3, 4, 5, 6, 7]);
+  }, 1050);
 });