feat: Add allValuesFrom function

api_guard/dist/types/index.d.ts

@@ -1,3 +1,5 @@
+export declare function allValuesFrom<T>(source: Observable<T>): Promise<T[]>;
+
 export declare const animationFrame: AnimationFrameScheduler;
 
 export declare function animationFrames(timestampProvider?: TimestampProvider): Observable<{

spec/allValuesFrom-spec.ts

@@ -0,0 +1,39 @@
+import { interval, allValuesFrom, EMPTY, throwError, of } from 'rxjs';
+import { expect } from 'chai';
+import { finalize, take } from 'rxjs/operators';
+
+describe('allValuesFrom', () => {
+  it('should emit all the values as a promise', async () => {
+    let finalized = false;
+    const source = interval(2).pipe(take(4), finalize(() => (finalized = true)));
+    const result = await allValuesFrom(source);
+    expect(result).to.deep.equal([0, 1, 2, 3]);
+    expect(finalized).to.be.true;
+  });
+
+  it('should produce empty arrays for empty observables', async () => {
+    const source = EMPTY;
+    const result = await allValuesFrom(source);
+    expect(result).to.be.empty;
+  });
+
+  it('should error for errored observables', async () => {
+    const source = throwError(() => new Error('blorp!'));
+    let error: any = null;
+    try {
+      await allValuesFrom(source);
+    } catch (err) {
+      error = err;
+    }
+    expect(error).to.be.an.instanceOf(Error);
+    expect(error.message).to.equal('blorp!');
+  });
+
+  it('should work with a synchronous observable', async () => {
+    let finalized = false;
+    const source = of('apples', 'bananas').pipe(finalize(() => (finalized = true)));
+    const result = await allValuesFrom(source);
+    expect(result).to.deep.equal(['apples', 'bananas']);
+    expect(finalized).to.be.true;
+  });
+});

src/index.ts

@@ -50,6 +50,7 @@ export { isObservable } from './internal/util/isObservable';
 /* Promise Conversion */
 export { lastValueFrom } from './internal/lastValueFrom';
 export { firstValueFrom } from './internal/firstValueFrom';
+export { allValuesFrom } from './internal/allValuesFrom';
 
 /* Error types */
 export { ArgumentOutOfRangeError } from './internal/util/ArgumentOutOfRangeError';

src/internal/allValuesFrom.ts

@@ -0,0 +1,54 @@
+import { Observable } from './Observable';
+import { firstValueFrom } from './firstValueFrom';
+import { toArray } from './operators/toArray';
+
+/**
+ * Converts an observable to a promise by subscribing to the observable,
+ * and returning a promise that will resolve as soon as it completes,
+ * producing an array of all the values that the observable emitted. (If the
+ * observable completes without emitting any values, the promise will resolve
+ * with an empty array.)
+ *
+ * If the observable stream emits an error, the returned promise will reject
+ * with that error.
+ *
+ * **WARNING**: Only use this with observables you *know* will complete. If
+ * the source observable does not complete, you will end up with a promise
+ * that is hung up, and potentially all the state of an async function hanging
+ * out in memory. To avoid this situation, look into adding something like
+ * {@link timeout}, {@link take}, {@link takeWhile}, or {@link takeUntil}
+ * amongst others.
+ *
+ * ## Example
+ *
+ * Wait for the first three values from a stream and emit them from a promise
+ * in an async function
+ *
+ * ```ts
+ * import { interval, take, firstValueFrom } from 'rxjs';
+ *
+ * async function execute() {
+ *   const source$ = interval(2000).pipe(take(3));
+ *   const numbers = await allValuesFrom(source$);
+ *   console.log(`The numbers are: ${ numbers.join(', ') }`);
+ * }
+ *
+ * execute();
+ *
+ * // Expected output:
+ * // 'The numbers are: 0, 1, 2'
+ * ```
+ *
+ * Note that this function is equivalent to piping the observable through
+ * {@link toArray} and then calling either {@link firstValueFrom} or
+ * {@link lastValueFrom} (which are equivalent in this case).
+ *
+ * @see {@link firstValueFrom}
+ * @see {@link lastValueFrom}
+ * @see {@link toArray}
+ *
+ * @param source the observable to convert to a promise
+ */
+export function allValuesFrom<T>(source: Observable<T>): Promise<T[]> {
+  return firstValueFrom(source.pipe(toArray()));
+}

src/internal/firstValueFrom.ts

@@ -49,6 +49,7 @@ export function firstValueFrom<T>(source: Observable<T>): Promise<T>;
  * ```
  *
  * @see {@link lastValueFrom}
+ * @see {@link allValuesFrom}
  *
  * @param source the observable to convert to a promise
  * @param config a configuration object to define the `defaultValue` to use if the source completes without emitting a value

src/internal/lastValueFrom.ts

@@ -47,6 +47,7 @@ export function lastValueFrom<T>(source: Observable<T>): Promise<T>;
  * ```
  *
  * @see {@link firstValueFrom}
+ * @see {@link allValuesFrom}
  *
  * @param source the observable to convert to a promise
  * @param config a configuration object to define the `defaultValue` to use if the source completes without emitting a value

src/internal/operators/toArray.ts

@@ -31,6 +31,8 @@ const arrReducer = (arr: any[], value: any) => (arr.push(value), arr);
  * // output: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  * ```
  *
+ * @see {@link allValuesFrom}
+ *
  * @return A function that returns an Observable that emits an array of items
  * emitted by the source Observable when source completes.
  */