Skip to content

Commit

Permalink
feat(concatMap): add higher-order lettable version of concatMap
Browse files Browse the repository at this point in the history
  • Loading branch information
benlesh committed Jun 15, 2017
1 parent f0f7b98 commit c4125ff
Show file tree
Hide file tree
Showing 4 changed files with 75 additions and 105 deletions.
4 changes: 2 additions & 2 deletions src/operator/concatMap.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { MergeMapOperator } from './mergeMap';
import { concatMap as higherOrderConcatMap } from '../operators';
import { Observable, ObservableInput } from '../Observable';

/* tslint:disable:max-line-length */
Expand Down Expand Up @@ -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);
}
102 changes: 0 additions & 102 deletions src/operator/mergeMap.ts
Original file line number Diff line number Diff line change
@@ -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 */
Expand Down Expand Up @@ -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();
}
}
}
72 changes: 72 additions & 0 deletions src/operators/concatMap.ts
Original file line number Diff line number Diff line change
@@ -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);
}
2 changes: 1 addition & 1 deletion src/operators/index.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
export { concatMap } from './concatMap';
export { filter } from './filter';

export { map } from './map';
export { mergeMap } from './mergeMap';

0 comments on commit c4125ff

Please sign in to comment.