From 1a1d3f43b39c8105d5ce2f9466cf5395ada750ac Mon Sep 17 00:00:00 2001 From: Alexander Harding Date: Mon, 18 May 2026 22:48:14 -0600 Subject: [PATCH 1/3] Improved the readability of the all operator as well as added memory optimizations. --- all/deno.json | 2 +- all/mod.test.ts | 31 +++++++++- all/mod.ts | 151 +++++++++++++++++++++++++----------------------- 3 files changed, 110 insertions(+), 74 deletions(-) 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..b1e708f 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 { Observable, Observer } from "@observable/core"; import { materialize, type ObserverNotification } from "@observable/materialize"; import { flat } from "@observable/flat"; import { defer } from "@observable/defer"; @@ -106,6 +106,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..cd27992 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,86 @@ 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} is empty to be evaluated by subsequent logic. + */ + let isEmpty = true; + /** + * 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 (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 pushed, it prevents the buffer from + // being cloned unnecessarily. + if (!Object.is(buffer[indexSnapshot], value)) { + // Update the buffer. + buffer[indexSnapshot] = value; + // Reset the buffer snapshot since it's now stale. + snapshot = undefined; + } + + if (receivedFirstValueCount === observables.size) { + observer.next(snapshot ??= Object.freeze(buffer.slice())); + } + }, + return() { + if (isEmpty || !activeInnerSubscriptions) observer.return(); + }, + throw: (value) => observer.throw(value), + }); + + if (observer.signal.aborted) return; + } }); } @@ -226,10 +227,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 }; } From a9de15dccab89592f8ef4ef24118944ee952041e Mon Sep 17 00:00:00 2001 From: Alexander Harding Date: Mon, 18 May 2026 22:50:18 -0600 Subject: [PATCH 2/3] Fixed linting --- all/mod.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/all/mod.test.ts b/all/mod.test.ts index b1e708f..22519f1 100644 --- a/all/mod.test.ts +++ b/all/mod.test.ts @@ -1,5 +1,5 @@ import { assertEquals, assertStrictEquals, assertThrows } from "@std/assert"; -import { Observable, 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"; From 2f92dc527c403fc96d58fabd62e958de3bdfa5dd Mon Sep 17 00:00:00 2001 From: Alexander Harding Date: Mon, 18 May 2026 23:22:48 -0600 Subject: [PATCH 3/3] Fixed a bug where first values could not be undefined. --- all/mod.test.ts | 18 ++++++++++++++++++ all/mod.ts | 17 ++++++----------- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/all/mod.test.ts b/all/mod.test.ts index 22519f1..11692db 100644 --- a/all/mod.test.ts +++ b/all/mod.test.ts @@ -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([]); diff --git a/all/mod.ts b/all/mod.ts index cd27992..1c25f67 100644 --- a/all/mod.ts +++ b/all/mod.ts @@ -182,9 +182,9 @@ export function all( for (const observable of observables) { /** - * Tracking if the {@linkcode observable} is empty to be evaluated by subsequent logic. + * Tracking if the {@linkcode observable} has pushed its first value. */ - let isEmpty = true; + let receivedFirstValue = false; /** * Tracking a snapshot of the {@linkcode index} at the time of evaluation. */ @@ -193,17 +193,12 @@ export function all( pipe(observable, finalize(() => --activeInnerSubscriptions)).subscribe({ signal: observer.signal, next(value) { - if (isEmpty) ++receivedFirstValueCount; + if (!receivedFirstValue || !Object.is(buffer[indexSnapshot], value)) { + if (!receivedFirstValue) ++receivedFirstValueCount; - isEmpty = false; + receivedFirstValue = true; - // Check for value equality and update the buffer only if it's different. - // Though this doesn't stop the snapshot from being pushed, it prevents the buffer from - // being cloned unnecessarily. - if (!Object.is(buffer[indexSnapshot], value)) { - // Update the buffer. buffer[indexSnapshot] = value; - // Reset the buffer snapshot since it's now stale. snapshot = undefined; } @@ -212,7 +207,7 @@ export function all( } }, return() { - if (isEmpty || !activeInnerSubscriptions) observer.return(); + if (!receivedFirstValue || !activeInnerSubscriptions) observer.return(); }, throw: (value) => observer.throw(value), });