feat(fetch): Making fetch happen (ReactiveX#4702)

* feat(fetch): Making fetch happen

- Adds `fromFetch` implementation that uses native `fetch`
- Adds tests and basic documentation
- DOES NOT polyfill fetch

* fixup! feat(fetch): Making fetch happen

docs_app/tools/transforms/angular-api-package/index.js

@@ -70,6 +70,7 @@ module.exports = new Package('angular-api', [basePackage, typeScriptPackage])
       'index.ts',
       'operators/index.ts',
       'ajax/index.ts',
+      'fetch/index.ts',
       'webSocket/index.ts',
       'testing/index.ts'
     ];

integration/systemjs/systemjs-compatibility-spec.js

@@ -6,6 +6,7 @@ System.config({
   packages: {
     'rxjs': {main: 'index.js', defaultExtension: 'js' },
     'rxjs/ajax': {main: 'index.js', defaultExtension: 'js' },
+    'rxjs/fetch': {main: 'index.js', defaultExtension: 'js' },
     'rxjs/operators': {main: 'index.js', defaultExtension: 'js' },
     'rxjs/testing': {main: 'index.js', defaultExtension: 'js' },
     'rxjs/webSocket': {main: 'index.js', defaultExtension: 'js' }
@@ -15,6 +16,7 @@ System.config({
 Promise.all([
   System.import('rxjs'),
   System.import('rxjs/ajax'),
+  System.import('rxjs/fetch'),
   System.import('rxjs/operators'),
   System.import('rxjs/testing'),
   System.import('rxjs/webSocket'),

spec/observables/dom/fetch-spec.ts

@@ -0,0 +1,196 @@
+import { fromFetch } from 'rxjs/fetch';
+import { expect } from 'chai';
+import { root } from '../../../src/internal/util/root';
+
+const OK_RESPONSE = {
+  ok: true,
+} as Response;
+
+function mockFetchImpl(input: string | Request, init?: RequestInit): Promise<Response> {
+  (mockFetchImpl as MockFetch).calls.push({ input, init });
+  return new Promise<any>((resolve, reject) => {
+    if (init.signal) {
+      init.signal.addEventListener('abort', () => {
+        reject(new MockDOMException());
+      });
+    }
+    return Promise.resolve(null).then(() => {
+      resolve((mockFetchImpl as any).respondWith);
+    });
+  });
+}
+(mockFetchImpl as MockFetch).reset = function (this: any) {
+  this.calls = [] as any[];
+  this.respondWith = OK_RESPONSE;
+};
+(mockFetchImpl as MockFetch).reset();
+
+const mockFetch: MockFetch = mockFetchImpl as MockFetch;
+
+class MockDOMException {}
+
+class MockAbortController {
+  readonly signal = new MockAbortSignal();
+
+  abort() {
+    this.signal._signal();
+  }
+
+  constructor() {
+    MockAbortController.created++;
+  }
+
+  static created = 0;
+
+  static reset() {
+    MockAbortController.created = 0;
+  }
+}
+
+class MockAbortSignal {
+  private _listeners: Function[] = [];
+
+  aborted = false;
+
+  addEventListener(name: 'abort', handler: Function) {
+    this._listeners.push(handler);
+  }
+
+  removeEventListener(name: 'abort', handler: Function) {
+    const index = this._listeners.indexOf(handler);
+    if (index >= 0) {
+      this._listeners.splice(index, 1);
+    }
+  }
+
+  _signal() {
+    this.aborted = true;
+    while (this._listeners.length > 0) {
+      this._listeners.shift()();
+    }
+  }
+}
+
+interface MockFetch {
+  (input: string | Request, init?: RequestInit): Promise<Response>;
+  calls: { input: string | Request, init: RequestInit | undefined }[];
+  reset(): void;
+  respondWith: Response;
+}
+
+describe('fromFetch', () => {
+  let _fetch: typeof fetch;
+  let _AbortController: AbortController;
+
+  beforeEach(() => {
+    mockFetch.reset();
+    if (root.fetch) {
+      _fetch = root.fetch;
+    }
+    root.fetch = mockFetch;
+
+    MockAbortController.reset();
+    if (root.AbortController) {
+      _AbortController = root.AbortController;
+    }
+    root.AbortController = MockAbortController;
+  });
+
+  afterEach(() => {
+    root.fetch = _fetch;
+    root.AbortController = _AbortController;
+  });
+
+  it('should exist', () => {
+    expect(fromFetch).to.be.a('function');
+  });
+
+  it('should fetch', done => {
+    const fetch$ = fromFetch('/foo');
+    expect(mockFetch.calls.length).to.equal(0);
+    expect(MockAbortController.created).to.equal(0);
+
+    fetch$.subscribe({
+      next: response => {
+        expect(response).to.equal(OK_RESPONSE);
+      },
+      error: done,
+      complete: done,
+    });
+
+    expect(MockAbortController.created).to.equal(1);
+    expect(mockFetch.calls.length).to.equal(1);
+    expect(mockFetch.calls[0].input).to.equal('/foo');
+    expect(mockFetch.calls[0].init.signal).not.to.be.undefined;
+    expect(mockFetch.calls[0].init.signal.aborted).to.be.false;
+  });
+
+  it('should handle Response that is not `ok`', done => {
+    mockFetch.respondWith = {
+      ok: false,
+      status: 400,
+      body: 'Bad stuff here'
+    } as any as Response;
+
+    const fetch$ = fromFetch('/foo');
+    expect(mockFetch.calls.length).to.equal(0);
+    expect(MockAbortController.created).to.equal(0);
+
+    fetch$.subscribe({
+      next: response => {
+        expect(response).to.equal(mockFetch.respondWith);
+      },
+      complete: done,
+      error: done
+    });
+
+    expect(MockAbortController.created).to.equal(1);
+    expect(mockFetch.calls.length).to.equal(1);
+    expect(mockFetch.calls[0].input).to.equal('/foo');
+    expect(mockFetch.calls[0].init.signal).not.to.be.undefined;
+    expect(mockFetch.calls[0].init.signal.aborted).to.be.false;
+  });
+
+  it('should abort when unsubscribed', () => {
+    const fetch$ = fromFetch('/foo');
+    expect(mockFetch.calls.length).to.equal(0);
+    expect(MockAbortController.created).to.equal(0);
+    const subscription = fetch$.subscribe();
+
+    expect(MockAbortController.created).to.equal(1);
+    expect(mockFetch.calls.length).to.equal(1);
+    expect(mockFetch.calls[0].input).to.equal('/foo');
+    expect(mockFetch.calls[0].init.signal).not.to.be.undefined;
+    expect(mockFetch.calls[0].init.signal.aborted).to.be.false;
+
+    subscription.unsubscribe();
+    expect(mockFetch.calls[0].init.signal.aborted).to.be.true;
+  });
+
+  it('should allow passing of init object', done => {
+    const myInit = {};
+    const fetch$ = fromFetch('/foo', myInit);
+    fetch$.subscribe({
+      error: done,
+      complete: done,
+    });
+    expect(mockFetch.calls[0].init).to.equal(myInit);
+    expect(mockFetch.calls[0].init.signal).not.to.be.undefined;
+  });
+
+  it('should treat passed signals as a cancellation token which triggers an error', done => {
+    const controller = new MockAbortController();
+    const signal = controller.signal as any;
+    const fetch$ = fromFetch('/foo', { signal });
+    const subscription = fetch$.subscribe({
+      error: err => {
+        expect(err).to.be.instanceof(MockDOMException);
+        done();
+      }
+    });
+    controller.abort();
+    expect(mockFetch.calls[0].init.signal.aborted).to.be.true;
+    // The subscription will not be closed until the error fires when the promise resolves.
+    expect(subscription.closed).to.be.false;
+  });
+});

spec/support/default.opts

@@ -7,7 +7,7 @@
 --reporter dot
 
 --check-leaks
---globals WebSocket,FormData,XDomainRequest,ActiveXObject
+--globals WebSocket,FormData,XDomainRequest,ActiveXObject,fetch,AbortController
 
 --recursive
 --timeout 5000

src/fetch/index.ts

@@ -0,0 +1 @@
+export { fromFetch } from '../internal/observable/dom/fetch';

src/fetch/package.json

@@ -0,0 +1,8 @@
+{
+  "name": "rxjs/fetch",
+  "typings": "./index.d.ts",
+  "main": "./index.js",
+  "module": "../_esm5/fetch/index.js",
+  "es2015": "../_esm2015/fetch/index.js",
+  "sideEffects": false
+}

src/internal/observable/dom/fetch.ts

@@ -0,0 +1,90 @@
+import { Observable } from '../../Observable';
+
+/**
+ * Uses [the Fetch API](https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API) to
+ * make an HTTP request.
+ *
+ * **WARNING** Parts of the fetch API are still experimental. `AbortController` is
+ * required for this implementation to work and use cancellation appropriately.
+ *
+ * Will automatically set up an internal [AbortController](https://developer.mozilla.org/en-US/docs/Web/API/AbortController)
+ * in order to teardown the internal `fetch` when the subscription tears down.
+ *
+ * If a `signal` is provided via the `init` argument, it will behave like it usually does with
+ * `fetch`. If the provided `signal` aborts, the error that `fetch` normally rejects with
+ * in that scenario will be emitted as an error from the observable.
+ *
+ * ### Basic Use
+ *
+ * ```ts
+ * import { of } from 'rxjs';
+ * import { fetch } from 'rxjs/fetch';
+ * import { switchMap, catchError } from 'rxjs/operators';
+ *
+ * const data$ = fetch('https://api.github.com/users?per_page=5').pipe(
+ *  switchMap(response => {
+ *    if(responose.ok) {
+ *      // OK return data
+ *      return response.json();
+ *    } else {
+ *      // Server is returning a status requiring the client to try something else.
+ *      return of({ error: true, message: `Error ${response.status}` });
+ *    }
+ *  }),
+ *  catchError(err => {
+ *    // Network or other error, handle appropriately
+ *    console.error(err);
+ *    return of({ error: true, message: error.message })
+ *  })
+ * );
+ *
+ * data$.subscribe({
+ *  next: result => console.log(result),
+ *  complete: () => console.log('done')
+ * })
+ * ```
+ *
+ * @param input The resource you would like to fetch. Can be a url or a request object.
+ * @param init A configuration object for the fetch.
+ * [See MDN for more details](https://developer.mozilla.org/en-US/docs/Web/API/WindowOrWorkerGlobalScope/fetch#Parameters)
+ * @returns An Observable, that when subscribed to performs an HTTP request using the native `fetch`
+ * function. The {@link Subscription} is tied to an `AbortController` for the the fetch.
+ */
+export function fromFetch(input: string | Request, init?: RequestInit): Observable<Response> {
+  return new Observable<Response>(subscriber => {
+    const controller = new AbortController();
+    const signal = controller.signal;
+    let outerSignalHandler: () => void;
+    let unsubscribed = false;
+
+    if (init) {
+      // If we a signal is provided, just have it teardown. It's a cancellation token, basically.
+      if (init.signal) {
+        outerSignalHandler = () => {
+          if (!signal.aborted) {
+            controller.abort();
+          }
+        };
+        init.signal.addEventListener('abort', outerSignalHandler);
+      }
+      init.signal = signal;
+    } else {
+      init = { signal };
+    }
+
+    fetch(input, init).then(response => {
+      subscriber.next(response);
+      subscriber.complete();
+    }).catch(err => {
+      if (!unsubscribed) {
+        // Only forward the error if it wasn't an abort.
+        subscriber.error(err);
+      }
+    });
+
+    return () => {
+      unsubscribed = true;
+      controller.abort();
+    };
+  });
+}

tools/make-umd-compat-bundle.js

@@ -8,6 +8,7 @@ rollupBundle({
     'rxjs/operators': 'dist-compat/esm5_for_rollup/src/operators/index.js',
     'rxjs/webSocket': 'dist-compat/esm5_for_rollup/src/webSocket/index.js',
     'rxjs/ajax': 'dist-compat/esm5_for_rollup/src/ajax/index.js',
+    'rxjs/fetch': 'dist-compat/esm5_for_rollup/src/fetch/index.js',
     'rxjs/internal-compatibility': 'dist-compat/esm5_for_rollup/src/internal-compatibility/index.js',
     'rxjs': 'dist-compat/esm5_for_rollup/src/index.js',
   },

tsconfig/tsconfig.base.json

@@ -13,6 +13,7 @@
     // entry-points
     "../src/index.ts",
     "../src/ajax/index.ts",
+    "../src/fetch/index.ts",
     "../src/operators/index.ts",
     "../src/testing/index.ts",
     "../src/webSocket/index.ts",

tsconfig/tsconfig.legacy-reexport.json

@@ -16,6 +16,7 @@
       "rxjs/internal-compatibility": ["../dist/typings/internal-compatibility/index"],
       "rxjs/testing": ["../dist/typings/testing/index"],
       "rxjs/ajax": ["../dist/typings/ajax/index"],
+      "rxjs/fetch": ["../dist/typings/fetch/index"],
       "rxjs/operators": ["../dist/typings/operators/index"],
       "rxjs/webSocket": ["../dist/typings/webSocket/index"],
       "rxjs-compat": ["../dist-compat/typings/compat/Rx"],