Feature: AbortSignal support first value from last value from

spec/firstValueFrom-spec.ts

@@ -1,4 +1,5 @@
-import { interval, firstValueFrom, EMPTY, EmptyError, throwError, of, Observable } from 'rxjs';
+/** @prettier */
+import { interval, firstValueFrom, EMPTY, EmptyError, throwError, of, Observable, NEVER, AbortError } from 'rxjs';
 import { expect } from 'chai';
 import { finalize } from 'rxjs/operators';
 
@@ -61,7 +62,7 @@ describe('firstValueFrom', () => {
 
   it('should stop listening to a synchronous observable when resolved', async () => {
     const sideEffects: number[] = [];
-    const synchronousObservable = new Observable<number>(subscriber => {
+    const synchronousObservable = new Observable<number>((subscriber) => {
       // This will check to see if the subscriber was closed on each loop
       // when the unsubscribe hits (from the `take`), it should be closed
       for (let i = 0; !subscriber.closed && i < 10; i++) {
@@ -70,7 +71,43 @@ describe('firstValueFrom', () => {
       }
     });
 
-    const result = await firstValueFrom(synchronousObservable);
+    await firstValueFrom(synchronousObservable);
     expect(sideEffects).to.deep.equal([0]);
   });
+
+  if (typeof AbortController === 'function') {
+    it('should support abort signal', async () => {
+      const source = NEVER;
+      const ac = new AbortController();
+      const signal = ac.signal;
+      setTimeout(() => {
+        ac.abort();
+      });
+      let errorThrown: any;
+      try {
+        await firstValueFrom(source, { signal });
+      } catch (err) {
+        errorThrown = err;
+      }
+      expect(errorThrown).to.be.an.instanceOf(AbortError);
+    });
+
+    it('should support abort signal with a default value', async () => {
+      const source = NEVER;
+      const ac = new AbortController();
+      const signal = ac.signal;
+      setTimeout(() => {
+        ac.abort();
+      });
+      let errorThrown: any;
+      let result = 'not set';
+      try {
+        result = await firstValueFrom(source, { signal, defaultValue: 'bad' });
+      } catch (err) {
+        errorThrown = err;
+      }
+      expect(errorThrown).to.be.an.instanceOf(AbortError);
+      expect(result).to.equal('not set');
+    });
+  }
 });

spec/lastValueFrom-spec.ts

@@ -1,4 +1,4 @@
-import { interval, lastValueFrom, EMPTY, EmptyError, throwError, of } from 'rxjs';
+import { interval, lastValueFrom, EMPTY, EmptyError, throwError, of, AbortError, NEVER } from 'rxjs';
 import { expect } from 'chai';
 import { finalize, take } from 'rxjs/operators';
 
@@ -61,4 +61,41 @@ describe('lastValueFrom', () => {
     expect(result).to.equal('bananas');
     expect(finalized).to.be.true;
   });
+
+
+  if (typeof AbortController === 'function') {
+    it('should support abort signal', async () => {
+      const source = NEVER;
+      const ac = new AbortController();
+      const signal = ac.signal;
+      setTimeout(() => {
+        ac.abort();
+      });
+      let errorThrown: any;
+      try {
+        await lastValueFrom(source, { signal });
+      } catch (err) {
+        errorThrown = err;
+      }
+      expect(errorThrown).to.be.an.instanceOf(AbortError);
+    });
+
+    it('should support abort signal with a default value', async () => {
+      const source = NEVER;
+      const ac = new AbortController();
+      const signal = ac.signal;
+      setTimeout(() => {
+        ac.abort();
+      });
+      let errorThrown: any;
+      let result = 'not set';
+      try {
+        result = await lastValueFrom(source, { signal, defaultValue: 'bad' });
+      } catch (err) {
+        errorThrown = err;
+      }
+      expect(errorThrown).to.be.an.instanceOf(AbortError);
+      expect(result).to.equal('not set');
+    });
+  }
 });

src/index.ts

@@ -53,6 +53,7 @@ export { firstValueFrom } from './internal/firstValueFrom';
 
 /* Error types */
 export { ArgumentOutOfRangeError } from './internal/util/ArgumentOutOfRangeError';
+export { AbortError } from './internal/util/AbortError';
 export { EmptyError } from './internal/util/EmptyError';
 export { NotFoundError } from './internal/util/NotFoundError';
 export { ObjectUnsubscribedError } from './internal/util/ObjectUnsubscribedError';

src/internal/firstValueFrom.ts

@@ -1,12 +1,16 @@
 import { Observable } from './Observable';
 import { EmptyError } from './util/EmptyError';
 import { SafeSubscriber } from './Subscriber';
+import { AbortError } from './util/AbortError';
+import { linkSignalToSubscription } from './util/linkSignalToSubscription';
 
 export interface FirstValueFromConfig<T> {
-  defaultValue: T;
+  defaultValue?: T;
+  signal?: AbortSignal;
 }
 
-export function firstValueFrom<T, D>(source: Observable<T>, config: FirstValueFromConfig<D>): Promise<T | D>;
+export function firstValueFrom<T, D>(source: Observable<T>, config: { defaultValue: D; signal?: AbortSignal }): Promise<T | D>;
+export function firstValueFrom<T, D>(source: Observable<T>, config: { signal?: AbortSignal }): Promise<T>;
 export function firstValueFrom<T>(source: Observable<T>): Promise<T>;
 
 /**
@@ -54,22 +58,40 @@ export function firstValueFrom<T>(source: Observable<T>): Promise<T>;
  * @param config a configuration object to define the `defaultValue` to use if the source completes without emitting a value
  */
 export function firstValueFrom<T, D>(source: Observable<T>, config?: FirstValueFromConfig<D>): Promise<T | D> {
-  const hasConfig = typeof config === 'object';
   return new Promise<T | D>((resolve, reject) => {
+    // This is creating our subscriber, which is also our subscription.
     const subscriber = new SafeSubscriber<T>({
       next: (value) => {
-        resolve(value);
+        // We have a value, unsubscribe as soon as we can and then emit.
         subscriber.unsubscribe();
+        resolve(value);
       },
       error: reject,
       complete: () => {
-        if (hasConfig) {
-          resolve(config!.defaultValue);
+        // We should never hit complete if we have a value! This is because we're unsubscribing
+        // as soon as we get a value in `next`. Therefore any call that lands here means the
+        // promised value never arrived.
+        if (config && 'defaultValue' in config) {
+          // If they gave use a default value it, resolve the promise with that.
+          resolve(config.defaultValue!);
         } else {
+          // Otherwise, reject with an empty error because promises *must* resolve or reject.
+          // If we don't reject here, it will leave our promise hanging and any other promises
+          // that were built off of it will never resolve or reject, either.
           reject(new EmptyError());
         }
       },
     });
+    const signal = config?.signal;
+    if (signal) {
+      // The user provided an abort signal, wire it up.
+      linkSignalToSubscription(signal, subscriber, () => {
+        reject(new AbortError());
+      });
+    }
+
+    // Start our subscription. Notice we are not capturing the returned subscription
+    // because it's technically the same instance as the `subscriber` above.
     source.subscribe(subscriber);
   });
 }

src/internal/lastValueFrom.ts

@@ -1,11 +1,16 @@
 import { Observable } from './Observable';
+import { SafeSubscriber } from './Subscriber';
+import { AbortError } from './util/AbortError';
 import { EmptyError } from './util/EmptyError';
+import { linkSignalToSubscription } from './util/linkSignalToSubscription';
 
 export interface LastValueFromConfig<T> {
-  defaultValue: T;
+  defaultValue?: T;
+  signal?: AbortSignal;
 }
 
-export function lastValueFrom<T, D>(source: Observable<T>, config: LastValueFromConfig<D>): Promise<T | D>;
+export function lastValueFrom<T, D>(source: Observable<T>, config: { defaultValue: D; signal?: AbortSignal }): Promise<T | D>;
+export function lastValueFrom<T>(source: Observable<T>, config: { signal?: AbortSignal }): Promise<T>;
 export function lastValueFrom<T>(source: Observable<T>): Promise<T>;
 
 /**
@@ -53,25 +58,44 @@ export function lastValueFrom<T>(source: Observable<T>): Promise<T>;
  * @param config a configuration object to define the `defaultValue` to use if the source completes without emitting a value
  */
 export function lastValueFrom<T, D>(source: Observable<T>, config?: LastValueFromConfig<D>): Promise<T | D> {
-  const hasConfig = typeof config === 'object';
   return new Promise<T | D>((resolve, reject) => {
+    // We must track if we have a value or not, because if
+    // we don't, then the promised value never arrived.
     let _hasValue = false;
     let _value: T;
-    source.subscribe({
+    const subscriber = new SafeSubscriber<T>({
       next: (value) => {
+        // We have a value! The promise can resolve later.
         _value = value;
         _hasValue = true;
       },
       error: reject,
       complete: () => {
         if (_hasValue) {
+          // Happy path.
           resolve(_value);
-        } else if (hasConfig) {
-          resolve(config!.defaultValue);
+        } else if (config && 'defaultValue' in config) {
+          // The observable was empty, but we have a default value we'd like to emit.
+          resolve(config.defaultValue!);
         } else {
+          // if the observable is empty, and we don't have a default value, we'll reject with an EmptyError
+          // because promises _must_ resolve or reject. We cannot just leave this hanging.
           reject(new EmptyError());
         }
       },
     });
+
+    const signal = config?.signal;
+    if (signal) {
+      // The user provided an abort signal. Wire it up. The
+      // subscriber *is* the subscription.
+      linkSignalToSubscription(signal, subscriber, () => {
+        reject(new AbortError());
+      });
+    }
+
+    // Start the subscription. We are not keeping the subscription returned
+    // because it's technically the same instance as the subscriber.
+    source.subscribe(subscriber);
   });
 }

src/internal/util/AbortError.ts

@@ -0,0 +1,28 @@
+import { createErrorClass } from './createErrorClass';
+
+export interface AbortError extends Error {}
+
+export interface AbortErrorCtor {
+  /**
+   * @deprecated Internal implementation detail. Do not construct error instances.
+   * Cannot be tagged as internal: https://github.com/ReactiveX/rxjs/issues/6269
+   */
+  new (): AbortError;
+}
+
+/**
+ * An error thrown when an abort signal is received and causes a promise to reject.
+ *
+ * @see {@link firstValueFrom}
+ * @see {@link lastValueFrom}
+ *
+ * @class AbortError
+ */
+export const AbortError: AbortErrorCtor = createErrorClass(
+  (_super) =>
+    function AbortErrorImpl(this: any) {
+      _super(this);
+      this.name = 'AbortError';
+      this.message = 'Aborted by AbortSignal';
+    }
+);

src/internal/util/linkSignalToSubscription.ts

@@ -0,0 +1,12 @@
+import { Subscription } from '../Subscription';
+
+export function linkSignalToSubscription(signal: AbortSignal, subscription: Subscription, onAbort: () => void) {
+  const handler = () => {
+    subscription.unsubscribe();
+    onAbort();
+  };
+  signal.addEventListener('abort', handler, { once: true });
+  subscription.add(() => {
+    signal.removeEventListener('abort', handler);
+  });
+}