feat: add AsyncIteratorChainableSupplier

src/iterator/AsyncIterator.ts

@@ -0,0 +1,17 @@
+import { CustomError } from "universe-log";
+
+export interface AsyncIterator<T> {
+    next(value?: any): Promise<IteratorResult<T>>;
+}
+
+export namespace AsyncIterator {
+    export class AsyncIteratorError extends CustomError {
+        public static iteratorAlreadyDoneError(msg?: string): AsyncIteratorError {
+            return new AsyncIteratorError(`This iterator was already done ${msg ? ": " + msg : "."}`);
+        }
+
+        public constructor(message?: string, cause?: Error) {
+            super(message, cause);
+        }
+    }
+}

src/iterator/AsyncIteratorChainableSupplier.mock.test.ts

@@ -0,0 +1,41 @@
+import { ChainableSupplier } from "../chainable/ChainableSupplier";
+import { SimpleTaker } from "../chainable/SimpleTaker";
+
+import { AsyncIterator } from "./AsyncIterator";
+
+export namespace mock {
+    export interface SampleObject {
+        v: number;
+    }
+
+    export class AsyncIteratorMock<T> implements AsyncIterator<T> {
+        private values: T[];
+
+        public constructor(values: T[]) {
+            this.values = values;
+        }
+
+        public async next(): Promise<IteratorResult<T>> {
+            const shifted = this.values.shift();
+            if (shifted) {
+                return { value: shifted, done: this.values.length === 0 };
+            } else throw AsyncIterator.AsyncIteratorError.iteratorAlreadyDoneError();
+        }
+    }
+
+    export async function takeElemsFromSupplier<T>(
+        supplier: ChainableSupplier<T, any>,
+        takeCount: number = -1,
+    ): Promise<T[]> {
+        const takenElems: T[] = [];
+        supplier.chain(
+            new SimpleTaker<T>(elem => {
+                takenElems.push(elem);
+                const takeNext = takeCount > 0 ? takenElems.length < takeCount : true;
+                return takeNext;
+            }),
+        );
+        await supplier.start();
+        return takenElems;
+    }
+}

src/iterator/AsyncIteratorChainableSupplier.spec.test.ts

@@ -0,0 +1,114 @@
+import { expect } from "chai";
+import * as _ from "lodash";
+import "mocha";
+import * as sinon from "sinon";
+
+import { SimpleTaker } from "../chainable/SimpleTaker";
+import { Log } from "../Log";
+
+import { AsyncIterator } from "./AsyncIterator";
+import { AsyncIteratorChainableSupplier } from "./AsyncIteratorChainableSupplier";
+import { mock } from "./AsyncIteratorChainableSupplier.mock.test";
+
+Log.log().initialize();
+
+describe.only("AsyncIteratorChainableSupplier", function() {
+    it("gives all elements", async () => {
+        const iterableValues: mock.SampleObject[] = _.range(0, 20).map(i => ({ v: i }));
+        const iteratorMock = new mock.AsyncIteratorMock<mock.SampleObject>(_.cloneDeep(iterableValues));
+        const iteratorSupplier = new AsyncIteratorChainableSupplier(iteratorMock);
+        const takenValues = await mock.takeElemsFromSupplier(iteratorSupplier);
+        expect(takenValues).to.be.deep.equal(iterableValues);
+    });
+
+    it("gives error when iterator.next throws", async () => {
+        const iteratorMock: AsyncIterator<mock.SampleObject> = {
+            next(): Promise<IteratorResult<mock.SampleObject>> {
+                throw new Error("Sample error");
+            },
+        };
+        const iteratorSupplier = new AsyncIteratorChainableSupplier(iteratorMock);
+        const foundErrors: Error[] = [];
+        try {
+            await iteratorSupplier
+                .branch(me =>
+                    me.chain(new SimpleTaker<mock.SampleObject>(elem => true)).catch(error => {
+                        foundErrors.push(error);
+                        return false;
+                    }),
+                )
+                .start();
+            expect.fail("Should throw");
+        } catch (error) {
+            expect(error)
+                .to.haveOwnProperty("message")
+                .that.is.equal("Sample error");
+        }
+        expect(foundErrors)
+            .to.be.an("array")
+            .with.length(1);
+        expect(foundErrors[0])
+            .to.haveOwnProperty("message")
+            .that.is.equal("Sample error");
+    });
+
+    it("gives error when give throws", async () => {
+        const iterableValues: mock.SampleObject[] = _.range(0, 20).map(i => ({ v: i }));
+        const iteratorMock = new mock.AsyncIteratorMock<mock.SampleObject>(iterableValues);
+        const iteratorSupplier = new AsyncIteratorChainableSupplier(iteratorMock);
+        const foundErrors: Error[] = [];
+        await iteratorSupplier
+            .branch(me =>
+                me
+                    .chain(
+                        new SimpleTaker<mock.SampleObject>(elem => {
+                            throw new Error("Sample error");
+                        }),
+                    )
+                    .catch(error => {
+                        foundErrors.push(error);
+                        return true;
+                    }),
+            )
+            .start();
+        expect(foundErrors)
+            .to.be.an("array")
+            .with.length(1);
+        expect(foundErrors[0])
+            .to.haveOwnProperty("message")
+            .that.is.equal("Sample error");
+    });
+
+    it("stops iterating when taker does not want more", async () => {
+        const iterableValues: mock.SampleObject[] = _.range(0, 20).map(i => ({ v: i }));
+        const iteratorMock = new mock.AsyncIteratorMock<mock.SampleObject>(iterableValues);
+        const nextSpy = sinon.spy(iteratorMock, "next");
+        const iteratorSupplier = new AsyncIteratorChainableSupplier(iteratorMock);
+
+        await iteratorSupplier
+            .branch(me =>
+                me
+                    .chain(
+                        new SimpleTaker<mock.SampleObject>(elem => {
+                            return false;
+                        }),
+                    )
+                    .catch(error => {
+                        return false;
+                    }),
+            )
+            .start();
+        expect(nextSpy.callCount).to.be.equal(1);
+    });
+
+    it("stops iterating when iterator returns done", async () => {
+        const iterableValues: mock.SampleObject[] = _.range(0, 20).map(i => ({ v: i }));
+        const iteratorMock = new mock.AsyncIteratorMock<mock.SampleObject>(iterableValues);
+
+        const nextSpy = sinon.spy(iteratorMock.next);
+
+        const iteratorSupplier = new AsyncIteratorChainableSupplier(iteratorMock);
+        await mock.takeElemsFromSupplier(iteratorSupplier);
+        expect(nextSpy.callCount).to.be.equal(iterableValues.length);
+    });
+});

src/iterator/AsyncIteratorChainableSupplier.ts

@@ -0,0 +1,39 @@
+import { ChainableSupplier } from "../chainable/ChainableSupplier";
+
+import { AsyncIterator } from "./AsyncIterator";
+
+export class AsyncIteratorChainableSupplier<T> extends ChainableSupplier<T, AsyncIteratorChainableSupplier<T>> {
+    private iterator: AsyncIterator<T>;
+    private done: boolean = false;
+
+    constructor(iterator: AsyncIterator<T>) {
+        super();
+
+        this.iterator = iterator;
+    }
+
+    public async start(): Promise<void> {
+        while (!this.done) {
+            const { done } = await this.next();
+            this.done = done;
+        }
+    }
+
+    protected me(): AsyncIteratorChainableSupplier<T> {
+        return this;
+    }
+
+    private async next(): Promise<{ done: boolean }> {
+        try {
+            const { value, done } = await this.iterator.next();
+            const takerWantsMore = this.give(undefined, value);
+            return { done: done || !takerWantsMore };
+        } catch (error) {
+            const takerWantsMore = this.give(error, undefined);
+            if (!takerWantsMore) {
+                throw error;
+            }
+            return { done: !takerWantsMore };
+        }
+    }
+}