sync

src/async/impl.ts

@@ -6,10 +6,10 @@ import { getRandomElementAsync } from "../random/index.js";
 import { AsyncRandomOptions } from "../random/types.js";
 import { selectionSorter } from "../sorting.js";
 import { Sequence } from "../sync/types.js";
-import { MaybeAsyncAnyPredicate, MaybeAsyncConverter, MaybeAsyncEquater, MaybeAsyncBiConverter, MaybeAsyncAccumulator, MaybeAsyncComparer, MaybeAsyncAction, MaybePromiseLike, MaybeAsyncGenerator, MaybeAsyncSequence } from "../types.js";
+import { MaybeAsyncAnyPredicate, MaybeAsyncConverter, MaybeAsyncEquater, MaybeAsyncBiConverter, MaybeAsyncAccumulator, MaybeAsyncComparer, MaybeAsyncAction, MaybePromiseLike, MaybeAsyncGenerator, MaybeAsyncSequence, MaybePromise } from "../types.js";
 import { strictEquals, identity, operatorCompare, defaultArrayComparer, combineAsyncComparers, asAsyncIterable } from "../utils.js";
 import { array, empty, wrap } from "./index.js";
-import { AsyncSequence, GroupedAsyncSequence, OrderedAsyncSequence } from "./types.js";
+import { AsyncSequence, AsyncSequencePipeline, GroupedAsyncSequence, OrderedAsyncSequence } from "./types.js";
 
 export class AsyncSequenceMarker { }
 
@@ -20,7 +20,7 @@ export abstract class BaseAsyncSequence<TElement> extends AsyncSequenceMarker im
 
   abstract iterator(): AsyncIterator<TElement>;
 
-  apply<TResult>(pipeline: (sequence: AsyncSequence<TElement>) => TResult): TResult {
+  apply<TResult>(pipeline: AsyncSequencePipeline<TElement, TResult>): MaybePromise<TResult> {
     return pipeline(this);
   }
 
@@ -397,7 +397,7 @@ export abstract class BaseAsyncSequence<TElement> extends AsyncSequenceMarker im
     let next = await iterator.next();
 
     if (next.done) {
-      throw new Error("Sequence contains no element.")
+      throw new Error("Sequence contains no element.");
     }
 
     if (!selector) {
@@ -620,15 +620,15 @@ export abstract class BaseAsyncSequence<TElement> extends AsyncSequenceMarker im
     return new ReversedAsyncSequence<TElement>(this);
   }
 
-  chunked(size: number, asArray?: false): AsyncSequence<AsyncSequence<TElement>>;
-  chunked(size: number, asArray: true): AsyncSequence<TElement[]>;
-  chunked(size: number, asArray?: boolean): AsyncSequence<AsyncSequence<TElement>> | AsyncSequence<TElement[]> {
+  chunked(size: number): AsyncSequence<AsyncSequence<TElement>>;
+  chunked<TResult>(size: number, transformer: AsyncSequencePipeline<TElement, TResult>): AsyncSequence<TResult>;
+  chunked<TResult>(size: number, transformer?: AsyncSequencePipeline<TElement, TResult>): AsyncSequence<AsyncSequence<TElement>> | AsyncSequence<TResult> {
     if (size <= 0) {
       throw new Error("Chunk size must be positive.");
     }
 
-    const sequence = new ChunkedAsyncSequence<TElement>(this, size);
-    return asArray ? sequence : sequence.select(array);
+    const result = new ChunkedAsyncSequence<TElement>(this, size);
+    return transformer ? result.select(transformer) : result;
   }
 
   async random(options?: AsyncRandomOptions<TElement> | undefined): Promise<TElement | undefined> {
@@ -721,10 +721,6 @@ export class DelegatedAsyncSequence<TElement> extends AsyncSequenceMarker implem
     return this.#sequence;
   }
 
-  protected set sequence(value: AsyncSequence<TElement>) {
-    this.#sequence = value;
-  }
-
   [Symbol.asyncIterator]() {
     return this.iterator();
   }
@@ -733,7 +729,7 @@ export class DelegatedAsyncSequence<TElement> extends AsyncSequenceMarker implem
     return this.#sequence.iterator();
   }
 
-  apply<TResult>(pipeline: (sequence: AsyncSequence<TElement>) => TResult) {
+  apply<TResult>(pipeline: AsyncSequencePipeline<TElement, TResult>) {
     return this.#sequence.apply(pipeline);
   }
 
@@ -978,10 +974,10 @@ export class DelegatedAsyncSequence<TElement> extends AsyncSequenceMarker implem
     return this.#sequence.reversed();
   }
 
-  chunked(size: number, asArray?: false): AsyncSequence<AsyncSequence<TElement>>;
-  chunked(size: number, asArray: true): AsyncSequence<TElement[]>;
-  chunked(size: number, asArray?: any): AsyncSequence<AsyncSequence<TElement>> | AsyncSequence<TElement[]> {
-    return this.#sequence.chunked(size, asArray);
+  chunked(size: number): AsyncSequence<AsyncSequence<TElement>>;
+  chunked<TResult>(size: number, transformer: AsyncSequencePipeline<TElement, TResult>): AsyncSequence<TResult>;
+  chunked(size: number, transformer?: any): any {
+    return this.#sequence.chunked(size, transformer);
   }
 
   random(options?: AsyncRandomOptions<TElement> | undefined): Promise<TElement | undefined> {
@@ -1186,9 +1182,9 @@ export class WrappedObjectAsync<T> extends BaseAsyncSequence<T> {
 }
 
 export class WrappedArrayAsync<T> extends BaseAsyncSequence<T> {
-  readonly #array: MaybePromiseLike<T>[];
+  readonly #array: ReadonlyArray<MaybePromiseLike<T>>;
 
-  constructor(array: MaybePromiseLike<T>[]) {
+  constructor(array: ReadonlyArray<MaybePromiseLike<T>>) {
     super();
 
     this.#array = array;
@@ -1694,6 +1690,10 @@ class AppendAsyncSequence<T> extends BaseAsyncSequence<T> {
     return n < 0 ? -1 : n + 1;
   }
 
+  override async maxCount() {
+    return await this.#sequence.maxCount() + 1;
+  }
+
   override async *iterator() {
     yield* this.#sequence;
     yield this.#obj;
@@ -1716,23 +1716,29 @@ class PrependAsyncSequence<T> extends BaseAsyncSequence<T> {
     return n < 0 ? -1 : n + 1;
   }
 
+  override async maxCount() {
+    return await this.#sequence.maxCount() + 1;
+  }
+
   override async *iterator() {
     yield this.#obj;
     yield* this.#sequence;
   }
 }
 
-class PeekAsyncSequence<T> extends DelegatedAsyncSequence<T> {
+class PeekAsyncSequence<T> extends BaseAsyncSequence<T> {
+  readonly #sequence: AsyncSequence<T>;
   readonly #action: MaybeAsyncAction<T>;
 
   constructor(sequence: AsyncSequence<T>, action: MaybeAsyncAction<T>) {
-    super(sequence);
+    super();
 
+    this.#sequence = sequence;
     this.#action = action;
   }
 
   override async *iterator() {
-    for await (const obj of this.sequence) {
+    for await (const obj of this.#sequence) {
       await this.#action(obj);
       yield obj;
     }
@@ -2023,7 +2029,7 @@ class GroupByAsyncSequence<TElement, TKey, TResult> extends BaseAsyncSequence<Gr
   }
 }
 
-class ChunkedAsyncSequence<T> extends BaseAsyncSequence<T[]> {
+class ChunkedAsyncSequence<T> extends BaseAsyncSequence<AsyncSequence<T>> {
   readonly #sequence: AsyncSequence<T>;
   readonly #size: number;
 
@@ -2046,13 +2052,13 @@ class ChunkedAsyncSequence<T> extends BaseAsyncSequence<T[]> {
       chunk.push(obj);
 
       if (chunk.length === this.#size) {
-        yield chunk;
+        yield array(chunk);
         chunk = [];
       }
     }
 
     if (chunk.length > 0) {
-      yield chunk;
+      yield array(chunk);
     }
   }
 }
@@ -2179,7 +2185,18 @@ class CacheAsyncSequence<T> extends BaseAsyncSequence<T> {
   }
 
   override async *iterator() {
-    yield* this.#cache ??= await this.#sequence.toArray();
+    if (this.#cache) {
+      yield* this.#cache;
+    } else {
+      const cache = [];
+
+      for await (const e of this.#sequence) {
+        cache.push(e);
+        yield e;
+      }
+
+      this.#cache = cache;
+    }
   }
 }