diff --git a/all/deno.json b/all/deno.json index 88011ef..db1cf45 100644 --- a/all/deno.json +++ b/all/deno.json @@ -1,6 +1,6 @@ { "name": "@observable/all", - "version": "0.19.1", + "version": "0.20.0", "license": "MIT", "exports": "./mod.ts", "publish": { "exclude": ["*.test.ts"] } diff --git a/all/mod.test.ts b/all/mod.test.ts index b55175d..11692db 100644 --- a/all/mod.test.ts +++ b/all/mod.test.ts @@ -1,5 +1,5 @@ import { assertEquals, assertStrictEquals, assertThrows } from "@std/assert"; -import { Observer } from "@observable/core"; +import { type Observable, Observer } from "@observable/core"; import { materialize, type ObserverNotification } from "@observable/materialize"; import { flat } from "@observable/flat"; import { defer } from "@observable/defer"; @@ -10,6 +10,7 @@ import { pipe } from "@observable/pipe"; import { all } from "./mod.ts"; import { ReplaySubject } from "@observable/replay-subject"; import { throwError } from "@observable/throw-error"; +import { of } from "@observable/of"; Deno.test( "all should combine multiple input's Observables that next and return synchronously", @@ -98,6 +99,23 @@ Deno.test("all should handle reentrancy", () => { ]); }); +Deno.test("all should handle undefined first values", () => { + // Arrange + const notifications: Array>> = []; + const observable = all([of(undefined), of(undefined), of(undefined)]); + + // Act + pipe(observable, materialize()).subscribe( + new Observer((notification) => notifications.push(notification)), + ); + + // Assert + assertEquals(notifications, [ + ["next", [undefined, undefined, undefined]], + ["return"], + ]); +}); + Deno.test("all should return empty when given an empty array", () => { // Arrange / Act const observable = all([]); @@ -106,6 +124,35 @@ Deno.test("all should return empty when given an empty array", () => { assertStrictEquals(observable, empty); }); +Deno.test("all should return empty when given an empty set", () => { + // Arrange / Act + const observable = all(new Set()); + + // Assert + assertStrictEquals(observable, empty); +}); + +Deno.test("all should be empty when given an empty iterable", () => { + // Arrange + const notifications: Array>> = []; + const iterable: Iterable> = { + [Symbol.iterator]: () => ({ + next: () => ({ done: true, value: undefined }), + return: () => ({ done: true, value: undefined }), + throw: () => ({ done: true, value: undefined }), + }), + }; + const observable = all(iterable); + + // Act + pipe(observable, materialize()).subscribe( + new Observer((notification) => notifications.push(notification)), + ); + + // Assert + assertEquals(notifications, [["return"]]); +}); + Deno.test("all should accept Set of observables", () => { // Arrange const notifications: Array>> = []; diff --git a/all/mod.ts b/all/mod.ts index daeb7da..1c25f67 100644 --- a/all/mod.ts +++ b/all/mod.ts @@ -1,16 +1,13 @@ -import { type Observable, Subject } from "@observable/core"; -import { defer } from "@observable/defer"; +import { Observable } from "@observable/core"; import { empty } from "@observable/empty"; -import { forOf } from "@observable/for-of"; import { pipe } from "@observable/pipe"; -import { tap } from "@observable/tap"; -import { map } from "@observable/map"; -import { mergeMap } from "@observable/merge-map"; -import { filter } from "@observable/filter"; -import { until } from "@observable/until"; import { finalize } from "@observable/finalize"; -import { catchError } from "@observable/catch-error"; -import { throwError } from "@observable/throw-error"; + +/** + * An iterable that has a known `size` _before_ iteration begins. + * @internal Do NOT export + */ +type BoundedIterable = Iterable & Readonly>; /** * [Pushes](https://jsr.io/@observable/core#push) the latest {@linkcode Values|values} from _all_ of the given @@ -142,82 +139,81 @@ export function all( observables: Iterable>, ): Observable>; export function all( - observables: Iterable>, + iterable: Iterable>, ): Observable> { if (!arguments.length) throw new TypeError("1 argument required but 0 present"); - if (!isIterable(observables)) throw new TypeError("Parameter 1 is not of type 'Iterable'"); + if (!isIterable(iterable)) throw new TypeError("Parameter 1 is not of type 'Iterable'"); - if (Array.isArray(observables) && !observables.length) return empty; + if (Array.isArray(iterable) && !iterable.length) return empty; + if (iterable instanceof Set && !iterable.size) return empty; - // Use defer so we do not start iterating until subscription, we get a fresh iteration for each subscription, - // and we get a fresh variable scope for each subscription. - return defer(() => { + // We could compose a few different operators to achieve the same result, but this is + // a more direct implementation that is easier to understand and reason about. + return new Observable((observer) => { /** - * Tracking the number of first values that have been received. + * The bounded {@linkcode iterable} which has a known `size` for subsequent logic. */ - let receivedFirstValueCount = 0; + const observables = bound(iterable); + + if (!observables.size) return observer.return(); + /** - * The normalized {@linkcode observables} which has a known length for subsequent logic. + * Tracking the number of first values that have been received. */ - const observableArray: ReadonlyArray> = Array.isArray(observables) - ? observables - : Array.from(observables); + let receivedFirstValueCount = 0; /** - * Tracking the expected number of first values that need to be received before the first snapshot is emitted. + * Tracking the number of active inner subscriptions. */ - const expectedFirstValueCount = observableArray.length; + let activeInnerSubscriptions = observables.size; /** - * Tracking a known list of buffered values, so we don't have to clone them while nexting to prevent reentrant behaviors. + * Tracking a known list of {@linkcode buffer|buffered} values, so we don't have to clone them while nexting to prevent reentrant behaviors. */ - let bufferSnapshot: ReadonlyArray | undefined; + let snapshot: ReadonlyArray | undefined; /** * Tracking the buffered values. */ const buffer: Array = []; /** - * The [notifier](https://jsr.io/@observable/core#notifier) that will tell the output - * [`Observable`](https://jsr.io/@observable/core/doc/~/Observable) to stop pushing values. + * Tracking the index of the current observable. */ - const stop = new Subject(); - - return pipe( - forOf(observableArray), - mergeMap((observable, index) => { - /** - * Tracking if the observable is empty to be evaluated by subsequent logic. - */ - let isEmpty = true; - return pipe( - observable, - tap((value) => { - if (isEmpty) receivedFirstValueCount++; - isEmpty = false; - - // Check for value equality and update the buffer only if it's different. - // Though this doesn't stop the snapshot from being emitted, it prevents the buffer from - // being cloned unnecessarily. - if (!Object.is(buffer[index], value)) { - // Update the buffer. - buffer[index] = value; - // Reset the buffer snapshot since it's now stale. - bufferSnapshot = undefined; - } - }), - catchError((value) => { - isEmpty = false; - return throwError(value); - }), - finalize(() => isEmpty && stop.next()), - ); - }), - filter(() => receivedFirstValueCount === expectedFirstValueCount), - // All first values have been received, we can cleanup the notifier. - tap(() => stop.return()), - map(() => bufferSnapshot ??= Object.freeze(buffer.slice())), - until(stop), - ); + let index = 0; + + for (const observable of observables) { + /** + * Tracking if the {@linkcode observable} has pushed its first value. + */ + let receivedFirstValue = false; + /** + * Tracking a snapshot of the {@linkcode index} at the time of evaluation. + */ + const indexSnapshot = index++; + + pipe(observable, finalize(() => --activeInnerSubscriptions)).subscribe({ + signal: observer.signal, + next(value) { + if (!receivedFirstValue || !Object.is(buffer[indexSnapshot], value)) { + if (!receivedFirstValue) ++receivedFirstValueCount; + + receivedFirstValue = true; + + buffer[indexSnapshot] = value; + snapshot = undefined; + } + + if (receivedFirstValueCount === observables.size) { + observer.next(snapshot ??= Object.freeze(buffer.slice())); + } + }, + return() { + if (!receivedFirstValue || !activeInnerSubscriptions) observer.return(); + }, + throw: (value) => observer.throw(value), + }); + + if (observer.signal.aborted) return; + } }); } @@ -226,10 +222,16 @@ export function all( * @internal Do NOT export */ function isIterable(value: unknown): value is Iterable { - if (!arguments.length) throw new TypeError("1 argument required but 0 present"); - return ( - (typeof value === "object" && value !== null) && - Symbol.iterator in value && - typeof value[Symbol.iterator] === "function" - ); + return typeof value === "object" && value !== null && Symbol.iterator in value && + typeof value[Symbol.iterator] === "function"; +} + +/** + * Converts the given {@linkcode value} to a _reliable_ {@linkcode BoundedIterable}. + * @internal Do NOT export + */ +function bound(value: Iterable): BoundedIterable { + if (value instanceof Set) return value; + const array = Array.isArray(value) ? value : Array.from(value); + return { [Symbol.iterator]: () => array[Symbol.iterator](), size: array.length }; }