Demystifying RxJS, Part II: MergeMap

02-05-demystifying-rxjs-mergemap.ts

function mergeMap<T, R>(
  project: (value: T, index: number) => Observable<R>
): OperatorFunction<T, R> {
  let currentIndex = 0;
  return source =>
    new Observable(obs => {
      const subscriptions = new Set<Subscription>();
      const sub = source.subscribe(
        x => {
          const projected = project(x, currentIndex++);
          const innerSub = projected.subscribe(
            px => obs.next(px),
            err => obs.error(err),
            () => {
              subscriptions.delete(innerSub);
            }
          );
          subscriptions.add(innerSub);
        },
        err => obs.error(err),
        () => {
          Array.from(subscriptions).forEach(s => s.unsubscribe());
          obs.complete();
        }
      );
      subscriptions.add(sub);
      return () => Array.from(subscriptions).forEach(s => s.unsubscribe());
    });
}