Async Iterables: Initial

index.d.ts

@@ -136,7 +136,7 @@ declare module 'fluent-iter' {
         groupBy<TKey, TElement>(keySelector: (item: TValue, index: number) => TKey, elementSelector: (item: TValue, index: number) => TElement): FluentIterable<IGrouping<TKey, TElement>>;
 
         groupBy<TKey, TElement, TResult>(keySelector: (item: TValue, index: number) => TKey, elementSelector: (item: TValue, index: number) => TElement,
-                                         resultCreator: (key: TKey, items: Iterable<TElement>) => TResult): FluentIterable<TResult>;
+                                         resultCreator: (key: TKey, items: FluentIterable<TElement>) => TResult): FluentIterable<TResult>;
 
         /**
          * Order iterable ascending by a key
@@ -162,7 +162,7 @@ declare module 'fluent-iter' {
         groupJoin<TInner, TKey, TResult>(joinIterable: Iterable<TInner>,
                                          sourceKeySelector: (item: TValue) => TKey,
                                          joinIterableKeySelector: (item: TInner, index: number) => TKey,
-                                         resultCreator: (outer: TValue, inner: TInner[]) => TResult): FluentIterable<TResult>;
+                                         resultCreator: (outer: TValue, inner: FluentIterable<TInner>) => TResult): FluentIterable<TResult>;
 
         /**
          * Do an inner join between current and external sequence. For each item of current sequence get a item from external sequence.
@@ -420,10 +420,40 @@ declare module 'fluent-iter' {
         isElementsEqual<TAnotherValue>(iterable: Iterable<TAnotherValue>, comparer: (a: TValue, b: TAnotherValue) => boolean): boolean;
     }
 
-    export interface IGrouping<TKey, TValue> extends Iterable<TValue> {
+    export interface IGrouping<TKey, TValue> extends FluentIterable<TValue> {
         key: TKey;
     }
 
+    export interface FluentIterableAsync<TValue> extends AsyncIterable<TValue> {
+        /**
+         * Filters the iterable using predicate function typed overload
+         * @param predicate
+         */
+        where<TSubValue extends TValue>(predicate: (item: TValue) => item is TSubValue): FluentIterableAsync<TSubValue>;
+
+        /**
+         * Filters the iterable using predicate function
+         * @param predicate
+         */
+        where(predicate: (item: TValue) => boolean): FluentIterableAsync<TValue>;
+        /**
+         * Maps the iterable items
+         * @param map map function
+         */
+        select<TOutput>(map: (item: TValue) => TOutput): FluentIterableAsync<TOutput>;
+
+        /**
+         * Take first N items from iterable
+         */
+        take(count: number): FluentIterableAsync<TValue>;
+
+        /**
+         * Return a promise to an array.
+         */
+        toArray(): Promise<TValue[]>;
+        toArray<TResult>(map: (item: TValue) => TResult): Promise<TResult[]>;
+    }
+
     export function from<TValue>(iterable: Iterable<TValue> | ArrayLike<TValue>): FluentIterable<TValue>;
     export function from<TValue extends {}, TKey extends keyof TValue>(value: TValue): FluentIterable<{ key: string, value: TValue[TKey] }>;
 
@@ -435,4 +465,7 @@ declare module 'fluent-iter' {
 
     export function fromObject<TValue extends {}, TKey extends keyof TValue>(value: TValue): FluentIterable<{ key: string, value: TValue[TKey] }>;
     export function fromObject<TValue extends {}, TKey extends keyof TValue, TResult>(value: TValue, resultCreator: (key: TKey, value: TValue[TKey]) => TResult): FluentIterable<TResult>;
+
+    export function fromEvent<TTarget extends EventTarget, TEvent extends keyof HTMLElementEventMap>(target: TTarget, event: TEvent): FluentIterableAsync<HTMLElementEventMap[TEvent]>;
+    export function fromTimer(interval: number, delay?: number): FluentIterableAsync<number>;
 }

index.ts

@@ -1,8 +1,12 @@
 export {
+    // sync
     fromIterable,
     fromObject,
     fromArrayLike,
     range,
     from,
     repeat,
+    // async
+    fromEvent,
+    fromTimer,
 } from "./src/index.ts";

src/creation-async.ts

@@ -0,0 +1,12 @@
+import type { FluentIterableAsync } from 'fluent-iter';
+import FluentAsync from "./fluent-async.js";
+import fromEventAsync from "./generators/from-event.ts";
+import fromTimerAsync from "./generators/from-timer.js";
+
+export function fromEvent<TTarget extends EventTarget, TEvent extends keyof HTMLElementEventMap>(target: TTarget, event: TEvent): FluentIterableAsync<HTMLElementEventMap[TEvent]> {
+    return new FluentAsync(fromEventAsync(target, event));
+}
+
+export function fromTimer(interval: number, delay?: number): FluentIterableAsync<number> {
+    return new FluentAsync(fromTimerAsync(interval, delay));
+}

src/creation.ts

@@ -43,6 +43,8 @@ export function from<TValue>(source: Iterable<TValue> | ArrayLike<TValue> | TVal
     return fromObject(source as object);
 }
 
+
+
 function isIterable<T>(o: any): o is Iterable<T> {
     const iterator = o[Symbol.iterator];
     return typeof iterator === 'function';

src/finalizers/to-array.ts

@@ -1,9 +1,21 @@
 import type {Mapper} from "../interfaces.ts";
 
-export default function toArrayCollector<T, R>(source: Iterable<T>, map?: Mapper<T, R>): T[] | R[] {
+export function toArrayCollector<T, R>(source: Iterable<T>, map?: Mapper<T, R>): T[] | R[] {
     if (!map) {
         return Array.from(source);
     } else {
         return Array.from(source).map(map);
     }
 }
+
+export async function toArrayAsyncCollector<T, R>(source: AsyncIterable<T>, map?: Mapper<T, R>): Promise<(T|R)[]> {
+    const result: (T|R)[] = [];
+    for await (const item of source) {
+        if (map) {
+            result.push(map(item));
+        } else {
+            result.push(item);
+        }
+    }
+    return result;
+}

src/fluent-async.ts

@@ -0,0 +1,34 @@
+import type { FluentIterableAsync } from 'fluent-iter';
+import type {Mapper, Predicate} from "./interfaces.js";
+import {whereAsyncIterator} from "./iterables/where.js";
+import {selectIteratorAsync} from "./iterables/select.js";
+import takeIteratorAsync, {takeIterator} from "./iterables/take.js";
+import {toArrayAsyncCollector} from "./finalizers/to-array.js";
+
+export default class FluentAsync<TValue> implements FluentIterableAsync<TValue> {
+    readonly #source: AsyncIterable<TValue>;
+
+    constructor(source: AsyncIterable<TValue>) {
+        this.#source = source;
+    }
+
+    where<TSubValue extends TValue>(predicate: (item: TValue) => item is TSubValue): FluentIterableAsync<TSubValue>;
+    where(predicate: Predicate<TValue>): FluentIterableAsync<TValue>;
+    where<TSubValue>(predicate: Predicate<TValue>): FluentIterableAsync<TValue> | FluentIterableAsync<TSubValue> {
+        return new FluentAsync(whereAsyncIterator(this, predicate));
+    }
+    select<TOutput>(map: Mapper<TValue, TOutput>): FluentIterableAsync<TOutput> {
+        return new FluentAsync(selectIteratorAsync(this, map));
+    }
+    take(count: number): FluentIterableAsync<TValue> {
+        return new FluentAsync(takeIteratorAsync(this, count));
+    }
+    toArray(): Promise<TValue[]>;
+    toArray<TResult>(map: Mapper<TValue, TResult>): Promise<TResult[]>;
+    toArray<TResult>(map?: Mapper<TValue, TResult>): Promise<(TValue|TResult)[]> {
+        return toArrayAsyncCollector(this, map);
+    }
+    [Symbol.asyncIterator](): AsyncIterator<TValue> {
+        return this.#source[Symbol.asyncIterator]();
+    }
+}

src/fluent.ts

@@ -1,9 +1,9 @@
-import whereIterator from "./iterables/where.ts";
-import selectIterator from "./iterables/select.ts";
+import { whereIterator } from "./iterables/where.ts";
+import { selectIterator } from "./iterables/select.ts";
 import selectManyIterator from "./iterables/select-many.ts";
-import takeIterator from "./iterables/take.ts";
+import { takeIterator } from "./iterables/take.ts";
 import skipIterator from "./iterables/skip.ts";
-import toArrayCollector from "./finalizers/to-array.ts";
+import { toArrayCollector } from "./finalizers/to-array.ts";
 import takeWhileIterator from "./iterables/take-while.ts";
 import skipWhileIterator from "./iterables/skip-while.ts";
 import takeLastIterator from "./iterables/take-last.ts";
@@ -38,7 +38,6 @@ import zipIterable from "./iterables/zip.js";
 import type {Action, Comparer, Mapper, Predicate} from "./interfaces.ts";
 import type { FluentIterable, IGrouping } from 'fluent-iter';
 
-
 export default class Fluent<TValue> implements FluentIterable<TValue> {
     readonly #source: Iterable<TValue>;
 
@@ -88,10 +87,10 @@ export default class Fluent<TValue> implements FluentIterable<TValue> {
     }
     groupBy<TKey>(keySelector: (item: TValue, index: number) => TKey): FluentIterable<IGrouping<TKey, TValue>>;
     groupBy<TKey, TElement>(keySelector: (item: TValue, index: number) => TKey, elementSelector: (item: TValue, index: number) => TElement): FluentIterable<IGrouping<TKey, TElement>>;
-    groupBy<TKey, TElement, TResult>(keySelector: (item: TValue, index: number) => TKey, elementSelector: (item: TValue, index: number) => TElement, resultCreator: (key: TKey, items: Iterable<TElement>) => TResult): FluentIterable<TResult>;
+    groupBy<TKey, TElement, TResult>(keySelector: (item: TValue, index: number) => TKey, elementSelector: (item: TValue, index: number) => TElement, resultCreator: (key: TKey, items: FluentIterable<TElement>) => TResult): FluentIterable<TResult>;
     groupBy<TKey, TElement, TResult>(keySelector: (item: TValue, index: number) => TKey,
                                      elementSelector?: (item: TValue, index: number) => TElement,
-                                     resultCreator?: (key: TKey, items: Iterable<TElement>) => TResult): FluentIterable<IGrouping<TKey, TValue> | IGrouping<TKey, TElement> | TResult> {
+                                     resultCreator?: (key: TKey, items: FluentIterable<TElement>) => TResult): FluentIterable<IGrouping<TKey, TValue> | IGrouping<TKey, TElement> | TResult> {
         return new Fluent(groupByIterator(this, keySelector, elementSelector, resultCreator));
     }
     orderBy<TKey>(keySelector: (item: TValue) => TKey, comparer?: Comparer<TKey>): FluentIterable<TValue> {
@@ -103,8 +102,8 @@ export default class Fluent<TValue> implements FluentIterable<TValue> {
     groupJoin<TInner, TKey, TResult>(joinIterable: Iterable<TInner>,
                                      sourceKeySelector: (item: TValue) => TKey,
                                      joinIterableKeySelector: (item: TInner, index: number) => TKey,
-                                     resultCreator: (outer: TValue, inner: TInner[]) => TResult): FluentIterable<TResult> {
-        return new Fluent(groupJoinIterator(this, joinIterable, sourceKeySelector, joinIterableKeySelector, resultCreator));
+                                     resultCreator: (outer: TValue, inner: FluentIterable<TInner> & TInner[]) => TResult): FluentIterable<TResult> {
+        return new Fluent(groupJoinIterator(this, joinIterable, sourceKeySelector, joinIterableKeySelector, resultCreator as any));
     }
     join(separator: string): string;
     join<TInner, TKey, TResult>(joinIterable: Iterable<TInner>,
@@ -244,3 +243,15 @@ export default class Fluent<TValue> implements FluentIterable<TValue> {
         return this.#source[Symbol.iterator]();
     }
 }
+
+export class Grouping<TKey, TValue> extends Fluent<TValue> implements IGrouping<TKey, TValue> {
+    readonly #key: TKey;
+    constructor(key: TKey, items: Iterable<TValue>) {
+        super(items);
+        this.#key = key;
+    }
+
+    public get key() {
+        return this.#key;
+    }
+}

src/generators/from-event.ts

@@ -0,0 +1,51 @@
+import {doneValue, iteratorResultCreator} from "../utils.ts";
+
+export default function fromEventAsync<TTarget extends EventTarget, TEvent extends keyof HTMLElementEventMap>(target: TTarget, event: TEvent): AsyncIterable<HTMLElementEventMap[TEvent]> & AsyncDisposable {
+    const eventQueue: HTMLElementEventMap[TEvent][] = [];
+    const resolverQueue: ((result: IteratorResult<HTMLElementEventMap[TEvent]>) => void)[] = [];
+
+    const eventHandler = (e: Event) => {
+        const eventValue = e as HTMLElementEventMap[TEvent];
+
+        const nextResolver = resolverQueue.shift();
+        if (nextResolver) {
+            nextResolver(iteratorResultCreator(eventValue));
+        } else {
+            eventQueue.push(eventValue);
+        }
+    };
+
+    target.addEventListener(event, eventHandler);
+    return {
+        [Symbol.asyncIterator]() {
+            return {
+                next(): Promise<IteratorResult<HTMLElementEventMap[TEvent]>> {
+                    const nextEvent = eventQueue.shift();
+                    if (nextEvent) {
+                        return Promise.resolve(iteratorResultCreator(nextEvent));
+                    }
+
+                    return new Promise((resolve) => {
+                        resolverQueue.push(resolve);
+                    });
+                },
+
+                return(): Promise<IteratorResult<HTMLElementEventMap[TEvent]>> {
+                    target.removeEventListener(event, eventHandler);
+                    resolverQueue.length = 0;
+                    return Promise.resolve(doneValue());
+                },
+                throw(error: any) {
+                    target.removeEventListener(event, eventHandler);
+                    resolverQueue.length = 0;
+                    return Promise.reject(error);
+                }
+            };
+        },
+        [Symbol.asyncDispose]: () => {
+            target.removeEventListener(event, eventHandler);
+            resolverQueue.length = 0;
+            return Promise.resolve();
+        }
+    };
+}

src/generators/from-timer.ts

@@ -0,0 +1,28 @@
+import {delay as sleep} from "../utils.ts";
+
+export default function fromTimerAsync(interval: number, delay?: number): AsyncIterable<number> & AsyncDisposable {
+    let done = false;
+    return {
+        [Symbol.asyncIterator]: async function* () {
+            let i = 0;
+            if (delay) {
+                await sleep(delay);
+                if (done) {
+                    return;
+                }
+                yield i++;
+            }
+            while (true) {
+                if (done) {
+                    break;
+                }
+                await sleep(interval);
+                yield i++;
+            }
+        },
+        [Symbol.asyncDispose]() {
+            done = true;
+            return Promise.resolve();
+        }
+    }
+}

src/index.ts

@@ -1 +1,2 @@
 export { fromIterable, fromObject, fromArrayLike, range, from, repeat } from './creation.ts';
+export { fromEvent, fromTimer } from './creation-async.ts';

src/iterables/group-join.ts

@@ -1,11 +1,13 @@
+import { FluentIterable } from "fluent-iter";
 import {createIterable, defaultElementSelector, group} from "../utils.ts";
+import {from} from "../creation.ts";
 
 export default function groupJoinIterator<TThis, TOther, TKey, TResult>(
     source: Iterable<TThis>,
     joinIterable: Iterable<TOther>,
     sourceKeySelector: (sourceItem: TThis) => TKey,
     joinIterableKeySelector: (otherItem: TOther, index: number) => TKey,
-    resultCreator: (first: TThis, second: TOther[]) => TResult): Iterable<TResult> {
+    resultCreator: (first: TThis, second: FluentIterable<TOther>) => TResult): Iterable<TResult> {
     return createIterable(() => groupJoinGenerator(source, joinIterable, sourceKeySelector, joinIterableKeySelector, resultCreator));
 }
 
@@ -14,11 +16,11 @@ function* groupJoinGenerator<TThis, TOther, TKey, TResult>(
     joinIterable: Iterable<TOther>,
     sourceKeySelector: (sourceItem: TThis) => TKey,
     joinIterableKeySelector: (otherItem: TOther, index: number) => TKey,
-    resultCreator: (first: TThis, second: TOther[]) => TResult): Generator<TResult> {
+    resultCreator: (first: TThis, second: FluentIterable<TOther>) => TResult): Generator<TResult> {
     const innerMap = group(joinIterable, joinIterableKeySelector, defaultElementSelector);
     for (const outerItem of source) {
         const key = sourceKeySelector(outerItem);
-        const innerItems = innerMap.get(key) || [];
+        const innerItems = innerMap.get(key) || from([]);
         yield resultCreator(outerItem, innerItems);
     }
 }

src/iterables/group.ts

@@ -1,42 +1,26 @@
-import type {IGrouping} from "fluent-iter";
-import {createIterable, getIterator, group} from "../utils.ts";
+import type {FluentIterable, IGrouping} from "fluent-iter";
+import {createIterable, group} from "../utils.ts";
+import {Grouping} from "../fluent.js";
 
 export function groupByIterator<TValue, TKey, TElement, TResult>(
     source: Iterable<TValue>,
     keySelector: (item: TValue, index: number) => TKey,
     elementSelector?: (item: TValue, index: number) => TElement,
-    resultCreator?: (key: TKey, items: Iterable<TElement>) => TResult): Iterable<IGrouping<TKey, TValue> | IGrouping<TKey, TElement> | TResult> {
+    resultCreator?: (key: TKey, items: FluentIterable<TElement>) => TResult): Iterable<IGrouping<TKey, TValue> | IGrouping<TKey, TElement> | TResult> {
     return createIterable(() => groupByGenerator(source, keySelector, elementSelector, resultCreator));
 }
 
 function* groupByGenerator<TValue, TKey, TElement=TValue, TResult=TValue>(
     source: Iterable<TValue>,
     keySelector: (item: TValue, index: number) => TKey,
     elementSelector?: (item: TValue, index: number) => TElement,
-    resultCreator?: (key: TKey, items: Iterable<TElement>) => TResult): Generator<IGrouping<TKey, TValue> | IGrouping<TKey, TElement> | TResult> {
+    resultCreator?: (key: TKey, items: FluentIterable<TElement>) => TResult): Generator<IGrouping<TKey, TValue> | IGrouping<TKey, TElement> | TResult> {
     const elementSelect: (item: TValue, index: number) => TElement = elementSelector ?? ((item) => item as unknown as TElement);
-    const resultCreate: (key: TKey, items: Iterable<TElement>) => TResult = resultCreator ??
+    const resultCreate: (key: TKey, items: FluentIterable<TElement>) => TResult = resultCreator ??
         ((key, items) => new Grouping(key, items) as unknown as TResult);
 
     const groups = group(source, keySelector, elementSelect);
     for (const [key, items] of groups.entries()) {
         yield resultCreate(key, items);
     }
 }
-
-export class Grouping<TKey, TValue> implements IGrouping<TKey, TValue> {
-    readonly #key: TKey;
-    readonly #items: Iterable<TValue>;
-    constructor(key: TKey, items: Iterable<TValue>) {
-        this.#key = key;
-        this.#items = items;
-    }
-
-    public get key() {
-        return this.#key;
-    }
-
-    [Symbol.iterator]() {
-        return getIterator(this.#items);
-    }
-}