Improved the readability of the all operator as well as added memory …

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"] }

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<ObserverNotification<ReadonlyArray<undefined>>> = [];
+  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<Observable>());
+
+  // Assert
+  assertStrictEquals(observable, empty);
+});
+
+Deno.test("all should be empty when given an empty iterable", () => {
+  // Arrange
+  const notifications: Array<ObserverNotification<ReadonlyArray<unknown>>> = [];
+  const iterable: Iterable<Observable<unknown>> = {
+    [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<ObserverNotification<ReadonlyArray<number>>> = [];

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<Value> = Iterable<Value> & Readonly<Record<"size", number>>;
 
 /**
  * [Pushes](https://jsr.io/@observable/core#push) the latest {@linkcode Values|values} from _all_ of the given
@@ -142,82 +139,81 @@ export function all<Value>(
   observables: Iterable<Observable<Value>>,
 ): Observable<ReadonlyArray<Value>>;
 export function all<Value>(
-  observables: Iterable<Observable<Value>>,
+  iterable: Iterable<Observable<Value>>,
 ): Observable<ReadonlyArray<Value>> {
   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<Observable<Value>> = 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<Value> | undefined;
+    let snapshot: ReadonlyArray<Value> | undefined;
     /**
      * Tracking the buffered values.
      */
     const buffer: Array<Value> = [];
 
     /**
-     * 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<void>();
-
-    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<Value>(
  * @internal Do NOT export
  */
 function isIterable(value: unknown): value is Iterable<unknown> {
-  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>(value: Iterable<Value>): BoundedIterable<Value> {
+  if (value instanceof Set) return value;
+  const array = Array.isArray(value) ? value : Array.from(value);
+  return { [Symbol.iterator]: () => array[Symbol.iterator](), size: array.length };
 }