artesanias/node_modules/rxjs/src/internal/observable/dom/fetch.ts

import { Observable } from '../../Observable';
import { Subscription } from '../../Subscription';
import { from } from '../../observable/from';
import { ObservableInput } from '../../types';

export function fromFetch<T>(
  input: string | Request,
  init: RequestInit & {
    selector: (response: Response) => ObservableInput<T>
  }
): Observable<T>;

export function fromFetch(
  input: string | Request,
  init?: RequestInit
): Observable<Response>;

/**
 * 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 { fromFetch } from 'rxjs/fetch';
 * import { switchMap, catchError } from 'rxjs/operators';
 *
 * const data$ = fromFetch('https://api.github.com/users?per_page=5').pipe(
 *  switchMap(response => {
 *    if (response.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: err.message })
 *  })
 * );
 *
 * data$.subscribe({
 *  next: result => console.log(result),
 *  complete: () => console.log('done')
 * });
 * ```
 *
 * ### Use with Chunked Transfer Encoding
 *
 * With HTTP responses that use [chunked transfer encoding](https://tools.ietf.org/html/rfc7230#section-3.3.1),
 * the promise returned by `fetch` will resolve as soon as the response's headers are
 * received.
 *
 * That means the `fromFetch` observable will emit a `Response` - and will
 * then complete - before the body is received. When one of the methods on the
 * `Response` - like `text()` or `json()` - is called, the returned promise will not
 * resolve until the entire body has been received. Unsubscribing from any observable
 * that uses the promise as an observable input will not abort the request.
 *
 * To facilitate aborting the retrieval of responses that use chunked transfer encoding,
 * a `selector` can be specified via the `init` parameter:
 *
 * ```ts
 * import { of } from 'rxjs';
 * import { fromFetch } from 'rxjs/fetch';
 *
 * const data$ = fromFetch('https://api.github.com/users?per_page=5', {
 *   selector: response => response.json()
 * });
 *
 * 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<T>(
  input: string | Request,
  initWithSelector: RequestInit & {
    selector?: (response: Response) => ObservableInput<T>
  } = {}
): Observable<Response | T> {
  const { selector, ...init } = initWithSelector;
  return new Observable<Response | T>(subscriber => {
    const controller = new AbortController();
    const signal = controller.signal;
    let abortable = true;
    let unsubscribed = false;

    const subscription = new Subscription();
    subscription.add(() => {
      unsubscribed = true;
      if (abortable) {
        controller.abort();
      }
    });

    let perSubscriberInit: RequestInit;
    if (init) {
      // If a signal is provided, just have it teardown. It's a cancellation token, basically.
      if (init.signal) {
        if (init.signal.aborted) {
          controller.abort();
        } else {
          const outerSignal = init.signal;
          const outerSignalHandler = () => {
            if (!signal.aborted) {
              controller.abort();
            }
          };
          outerSignal.addEventListener('abort', outerSignalHandler);
          subscription.add(() => outerSignal.removeEventListener('abort', outerSignalHandler));
        }
      }
      // init cannot be mutated or reassigned as it's closed over by the
      // subscriber callback and is shared between subscribers.
      perSubscriberInit = { ...init, signal };
    } else {
      perSubscriberInit = { signal };
    }

    fetch(input, perSubscriberInit).then(response => {
      if (selector) {
        subscription.add(from(selector(response)).subscribe(
          value => subscriber.next(value),
          err => {
            abortable = false;
            if (!unsubscribed) {
              // Only forward the error if it wasn't an abort.
              subscriber.error(err);
            }
          },
          () => {
            abortable = false;
            subscriber.complete();
          }
        ));
      } else {
        abortable = false;
        subscriber.next(response);
        subscriber.complete();
      }
    }).catch(err => {
      abortable = false;
      if (!unsubscribed) {
        // Only forward the error if it wasn't an abort.
        subscriber.error(err);
      }
    });

    return subscription;
  });
}