feat(concatMap): add higher-order lettable version of concatMap

src/operator/concatMap.ts

@@ -1,4 +1,4 @@
-import { MergeMapOperator } from './mergeMap';
+import { concatMap as higherOrderConcatMap } from '../operators';
 import { Observable, ObservableInput } from '../Observable';
 
 /* tslint:disable:max-line-length */
@@ -67,5 +67,5 @@ export function concatMap<T, I, R>(this: Observable<T>, project: (value: T, inde
  */
 export function concatMap<T, I, R>(this: Observable<T>, project: (value: T, index: number) =>  ObservableInput<I>,
                                    resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) {
-  return this.lift(new MergeMapOperator(project, resultSelector, 1));
+  return higherOrderConcatMap(project, resultSelector)(this);
 }

src/operator/mergeMap.ts

@@ -1,10 +1,4 @@
 import { Observable, ObservableInput } from '../Observable';
-import { Operator } from '../Operator';
-import { Subscriber } from '../Subscriber';
-import { Subscription } from '../Subscription';
-import { subscribeToResult } from '../util/subscribeToResult';
-import { OuterSubscriber } from '../OuterSubscriber';
-import { InnerSubscriber } from '../InnerSubscriber';
 import { mergeMap as higherOrderMergeMap } from '../operators';
 
 /* tslint:disable:max-line-length */
@@ -75,99 +69,3 @@ export function mergeMap<T, I, R>(this: Observable<T>, project: (value: T, index
                                   concurrent: number = Number.POSITIVE_INFINITY): Observable<I | R> {
   return higherOrderMergeMap(project, <any>resultSelector, concurrent)(this);
 }
-
-export class MergeMapOperator<T, I, R> implements Operator<T, I> {
-  constructor(private project: (value: T, index: number) => ObservableInput<I>,
-              private resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
-              private concurrent: number = Number.POSITIVE_INFINITY) {
-  }
-
-  call(observer: Subscriber<I>, source: any): any {
-    return source.subscribe(new MergeMapSubscriber(
-      observer, this.project, this.resultSelector, this.concurrent
-    ));
-  }
-}
-
-/**
- * We need this JSDoc comment for affecting ESDoc.
- * @ignore
- * @extends {Ignored}
- */
-export class MergeMapSubscriber<T, I, R> extends OuterSubscriber<T, I> {
-  private hasCompleted: boolean = false;
-  private buffer: T[] = [];
-  private active: number = 0;
-  protected index: number = 0;
-
-  constructor(destination: Subscriber<I>,
-              private project: (value: T, index: number) => ObservableInput<I>,
-              private resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R,
-              private concurrent: number = Number.POSITIVE_INFINITY) {
-    super(destination);
-  }
-
-  protected _next(value: T): void {
-    if (this.active < this.concurrent) {
-      this._tryNext(value);
-    } else {
-      this.buffer.push(value);
-    }
-  }
-
-  protected _tryNext(value: T) {
-    let result: ObservableInput<I>;
-    const index = this.index++;
-    try {
-      result = this.project(value, index);
-    } catch (err) {
-      this.destination.error(err);
-      return;
-    }
-    this.active++;
-    this._innerSub(result, value, index);
-  }
-
-  private _innerSub(ish: ObservableInput<I>, value: T, index: number): void {
-    this.add(subscribeToResult<T, I>(this, ish, value, index));
-  }
-
-  protected _complete(): void {
-    this.hasCompleted = true;
-    if (this.active === 0 && this.buffer.length === 0) {
-      this.destination.complete();
-    }
-  }
-
-  notifyNext(outerValue: T, innerValue: I,
-             outerIndex: number, innerIndex: number,
-             innerSub: InnerSubscriber<T, I>): void {
-    if (this.resultSelector) {
-      this._notifyResultSelector(outerValue, innerValue, outerIndex, innerIndex);
-    } else {
-      this.destination.next(innerValue);
-    }
-  }
-
-  private _notifyResultSelector(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) {
-    let result: R;
-    try {
-      result = this.resultSelector(outerValue, innerValue, outerIndex, innerIndex);
-    } catch (err) {
-      this.destination.error(err);
-      return;
-    }
-    this.destination.next(result);
-  }
-
-  notifyComplete(innerSub: Subscription): void {
-    const buffer = this.buffer;
-    this.remove(innerSub);
-    this.active--;
-    if (buffer.length > 0) {
-      this._next(buffer.shift());
-    } else if (this.active === 0 && this.hasCompleted) {
-      this.destination.complete();
-    }
-  }
-}

src/operators/concatMap.ts

@@ -0,0 +1,72 @@
+import { mergeMap } from './mergeMap';
+import { ObservableInput } from '../Observable';
+import { OperatorFunction } from '../interfaces';
+
+/* tslint:disable:max-line-length */
+export function concatMap<T, R>(project: (value: T, index: number) =>  ObservableInput<R>): OperatorFunction<T, R>;
+export function concatMap<T, I, R>(project: (value: T, index: number) =>  ObservableInput<I>, resultSelector: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R): OperatorFunction<T, R>;
+/* tslint:enable:max-line-length */
+
+/**
+ * Projects each source value to an Observable which is merged in the output
+ * Observable, in a serialized fashion waiting for each one to complete before
+ * merging the next.
+ *
+ * <span class="informal">Maps each value to an Observable, then flattens all of
+ * these inner Observables using {@link concatAll}.</span>
+ *
+ * <img src="./img/concatMap.png" width="100%">
+ *
+ * Returns an Observable that emits items based on applying a function that you
+ * supply to each item emitted by the source Observable, where that function
+ * returns an (so-called "inner") Observable. Each new inner Observable is
+ * concatenated with the previous inner Observable.
+ *
+ * __Warning:__ if source values arrive endlessly and faster than their
+ * corresponding inner Observables can complete, it will result in memory issues
+ * as inner Observables amass in an unbounded buffer waiting for their turn to
+ * be subscribed to.
+ *
+ * Note: `concatMap` is equivalent to `mergeMap` with concurrency parameter set
+ * to `1`.
+ *
+ * @example <caption>For each click event, tick every second from 0 to 3, with no concurrency</caption>
+ * var clicks = Rx.Observable.fromEvent(document, 'click');
+ * var result = clicks.concatMap(ev => Rx.Observable.interval(1000).take(4));
+ * result.subscribe(x => console.log(x));
+ *
+ * // Results in the following:
+ * // (results are not concurrent)
+ * // For every click on the "document" it will emit values 0 to 3 spaced
+ * // on a 1000ms interval
+ * // one click = 1000ms-> 0 -1000ms-> 1 -1000ms-> 2 -1000ms-> 3
+ *
+ * @see {@link concat}
+ * @see {@link concatAll}
+ * @see {@link concatMapTo}
+ * @see {@link exhaustMap}
+ * @see {@link mergeMap}
+ * @see {@link switchMap}
+ *
+ * @param {function(value: T, ?index: number): ObservableInput} project A function
+ * that, when applied to an item emitted by the source Observable, returns an
+ * Observable.
+ * @param {function(outerValue: T, innerValue: I, outerIndex: number, innerIndex: number): any} [resultSelector]
+ * A function to produce the value on the output Observable based on the values
+ * and the indices of the source (outer) emission and the inner Observable
+ * emission. The arguments passed to this function are:
+ * - `outerValue`: the value that came from the source
+ * - `innerValue`: the value that came from the projected Observable
+ * - `outerIndex`: the "index" of the value that came from the source
+ * - `innerIndex`: the "index" of the value from the projected Observable
+ * @return {Observable} An Observable that emits the result of applying the
+ * projection function (and the optional `resultSelector`) to each item emitted
+ * by the source Observable and taking values from each projected inner
+ * Observable sequentially.
+ * @method concatMap
+ * @owner Observable
+ */
+export function concatMap<T, I, R>(project: (value: T, index: number) =>  ObservableInput<I>,
+                                   resultSelector?: (outerValue: T, innerValue: I, outerIndex: number, innerIndex: number) => R) {
+  return mergeMap(project, resultSelector, 1);
+}

src/operators/index.ts

@@ -1,4 +1,4 @@
+export { concatMap } from './concatMap';
 export { filter } from './filter';
-
 export { map } from './map';
 export { mergeMap } from './mergeMap';