Feature: AbortSignal support

api_guard/dist/types/index.d.ts

@@ -261,6 +261,8 @@ export declare type InteropObservable<T> = {
 
 export declare function interval(period?: number, scheduler?: SchedulerLike): Observable<number>;
 
+export declare function isAbortError(value: any): boolean;
+
 export declare function isObservable<T>(obj: any): obj is Observable<T>;
 
 export declare function lastValueFrom<T>(source: Observable<T>): Promise<T>;
@@ -361,8 +363,8 @@ export declare class Observable<T> implements Subscribable<T> {
     constructor(subscribe?: (this: Observable<T>, subscriber: Subscriber<T>) => TeardownLogic);
     protected _subscribe(subscriber: Subscriber<any>): TeardownLogic;
     protected _trySubscribe(sink: Subscriber<T>): TeardownLogic;
-    forEach(next: (value: T) => void): Promise<void>;
-    forEach(next: (value: T) => void, promiseCtor: PromiseConstructorLike): Promise<void>;
+    forEach(nextHandler: (value: T) => void, signal?: AbortSignal): Promise<void>;
+    forEach(nextHandler: (value: T) => void, promiseCtor: PromiseConstructorLike): Promise<void>;
     protected lift<R>(operator?: Operator<T, R>): Observable<R>;
     pipe(): Observable<T>;
     pipe<A>(op1: OperatorFunction<T, A>): Observable<A>;
@@ -376,6 +378,7 @@ export declare class Observable<T> implements Subscribable<T> {
     pipe<A, B, C, D, E, F, G, H, I>(op1: OperatorFunction<T, A>, op2: OperatorFunction<A, B>, op3: OperatorFunction<B, C>, op4: OperatorFunction<C, D>, op5: OperatorFunction<D, E>, op6: OperatorFunction<E, F>, op7: OperatorFunction<F, G>, op8: OperatorFunction<G, H>, op9: OperatorFunction<H, I>): Observable<I>;
     pipe<A, B, C, D, E, F, G, H, I>(op1: OperatorFunction<T, A>, op2: OperatorFunction<A, B>, op3: OperatorFunction<B, C>, op4: OperatorFunction<C, D>, op5: OperatorFunction<D, E>, op6: OperatorFunction<E, F>, op7: OperatorFunction<F, G>, op8: OperatorFunction<G, H>, op9: OperatorFunction<H, I>, ...operations: OperatorFunction<any, any>[]): Observable<unknown>;
     subscribe(observer?: PartialObserver<T>): Subscription;
+    subscribe(observer: PartialObserver<T> | null | undefined, signal: AbortSignal | null | undefined): Subscription;
     subscribe(next: null | undefined, error: null | undefined, complete: () => void): Subscription;
     subscribe(next: null | undefined, error: (error: any) => void, complete?: () => void): Subscription;
     subscribe(next: (value: T) => void, error: null | undefined, complete: () => void): Subscription;

package.json

@@ -117,6 +117,7 @@
     "escape-string-regexp": "1.0.5",
     "eslint": "4.18.2",
     "eslint-plugin-jasmine": "^2.10.1",
+    "event-target-polyfill": "0.0.2",
     "fs-extra": "^8.1.0",
     "glob": "7.1.2",
     "google-closure-compiler-js": "20170218.0.0",
@@ -150,7 +151,8 @@
     "typedoc": "^0.17.8",
     "typescript": "~3.9.2",
     "validate-commit-msg": "2.14.0",
-    "webpack": "^4.31.0"
+    "webpack": "^4.31.0",
+    "yet-another-abortcontroller-polyfill": "0.0.3"
   },
   "files": [
     "dist/bundles",

spec/Observable-spec.ts

@@ -1,10 +1,12 @@
 import { expect } from 'chai';
 import * as sinon from 'sinon';
 import { Observer, TeardownLogic } from '../src/internal/types';
-import { Observable, config, Subscription, noop, Subscriber, Operator, NEVER, Subject, of, throwError, empty } from 'rxjs';
-import { map, multicast, refCount, filter, count, tap, combineLatest, concat, merge, race, zip, catchError } from 'rxjs/operators';
+import { Observable, config, Subscription, noop, Subscriber, Operator, NEVER, Subject, of, throwError, empty, interval } from 'rxjs';
+import { map, multicast, refCount, filter, count, tap, combineLatest, concat, merge, race, zip, catchError, mergeMap, finalize, mergeAll } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from './helpers/observableMatcher';
+import 'event-target-polyfill';
+import 'yet-another-abortcontroller-polyfill';
 
 function expectFullObserver(val: any) {
   expect(val).to.be.a('object');
@@ -65,6 +67,26 @@ describe('Observable', () => {
   });
 
   describe('forEach', () => {
+    it('should support AbortSignal', async () => {
+      const results: number[] = []
+      const ac = new AbortController();
+      try {
+      await of(1, 2, 3, 4, 5).forEach(
+        n => {
+          results.push(n);
+          if (n === 3) {
+            ac.abort();
+          }
+        },
+        ac.signal
+      ) 
+      } catch (err) {
+        expect(err.name).to.equal('AbortError');
+      }
+      expect(results).to.deep.equal([1, 2, 3]);
+    });
+
+
     it('should iterate and return a Promise', (done) => {
       const expected = [1, 2, 3];
       const result = of(1, 2, 3)
@@ -199,6 +221,84 @@ describe('Observable', () => {
     });
   });
 
+  describe('subscribe with abortSignal', () => {
+    it('should allow unsubscription with the abortSignal', (done) => {
+      const source = new Observable<number>(subscriber => {
+        let i = 0;
+        const id = setInterval(() => subscriber.next(i++));
+        return () => {
+          clearInterval(id);
+          expect(results).to.deep.equal([0, 1, 2, 3]);
+          expect(subscription.closed).to.be.true;
+          done();
+        }
+      });
+
+      const results: number[] = [];
+      const ac = new AbortController();
+      const subscription = source.subscribe({
+        next: n => {
+          results.push(n);
+          if (n === 3) {
+            ac.abort();
+          }
+        }
+      }, ac.signal);
+    });
+
+    it('should not subscribe if the abortSignal is already aborted', () => {
+      let called = false;
+      const source = new Observable(() => {
+        called = true;
+        throw new Error('should not be called');
+      });
+      const ac = new AbortController();
+      ac.abort();
+      const subscription = source.subscribe(undefined, ac.signal);
+      expect(called).to.be.false;
+      expect(subscription.closed).to.be.true;
+    });
+
+    it('should still chain the unsubscriptions', () => {
+      rxTestScheduler.run(({ hot, cold, expectObservable, expectSubscriptions, time }) => {
+        const inner1 = cold('    ----a----a-----a-|');
+        const inner2 = cold('        ----b----b------b---|');
+        const inner3 = cold('            ----c----c----c----c---|');
+        const source = hot('  ---a---b---c---|', { a: inner1, b: inner2, c: inner3 });
+        const sSubs = '       ^--------------!';
+        const i1Subs = '      ---^----------------!';
+        const i2Subs = '      -------^--------------!';
+        const i3Subs = '      -----------^----------!';
+        const abortAt = time('----------------------|')
+        const expected = '    -------a---ba--cb-a-c---';
+        const result = source.pipe(
+          mergeAll()
+        );
+
+        const ac = new AbortController();
+
+        const wrapperBecauseTestSchedulerDoesntSupportAbortYet = new Observable<string>(subscriber => {
+          return result.subscribe({
+            next: value => {
+              subscriber.next(value);
+            },
+            error: err => subscriber.error(err),
+            complete: () => subscriber.complete()
+          }, ac.signal);
+        });
+        rxTestScheduler.schedule(() => {
+          ac.abort();
+        }, abortAt);
+
+        expectObservable(wrapperBecauseTestSchedulerDoesntSupportAbortYet).toBe(expected);
+        expectSubscriptions(source.subscriptions).toBe(sSubs);
+        expectSubscriptions(inner1.subscriptions).toBe(i1Subs);
+        expectSubscriptions(inner2.subscriptions).toBe(i2Subs);
+        expectSubscriptions(inner3.subscriptions).toBe(i3Subs);
+      })
+    });
+  });
+
   describe('subscribe', () => {
     it('should be synchronous', () => {
       let subscribed = false;

src/index.ts

@@ -1,5 +1,5 @@
 /* Observable */
-export { Observable } from './internal/Observable';
+export { Observable, isAbortError } from './internal/Observable';
 export { ConnectableObservable } from './internal/observable/ConnectableObservable';
 export { GroupedObservable } from './internal/operators/groupBy';
 export { Operator } from './internal/Operator';