Promise.resolve（）但是Observables（RxJS5）
好了，为了更好地学习RxJS，我决定尝试创build一个自定义的Rx操作符。
所以这是一个简单的工作正常：
Rx.Observable.prototype.multiply = function (input) { const source = this; return Rx.Observable.create(function (obs) { return source.subscribe(function(val){ obs.next(input*val); }); }); }; 我们可以像这样使用它：
const obs = Rx.Observable.interval(1000) .multiply(4) .forEach(function (v) { console.log(v); });
然而，如果我们得到一些更复杂的东西，例如，如果我们的操作符采用一个函数而不是一个静态值。
Rx.Observable.prototype.handleFn = function (fn) { const source = this; return Rx.Observable.create(function (obs) { return source.subscribe(function(val){ obs.next(fn.call(obs,val)); }); }); };
上面的内容都很好， 但是如果我们需要处理从input函数返回的Rx.Observable，如下所示：
const obs = Rx.Observable.interval(1000) .handleFn(function(){ return Rx.Observable.timer(399); }) .forEach(function (v) { console.log(v); });
是否有某种Promise.resolve（），但为Observables，以便我可以解决Rx.Observable.timer（）的结果？ 将检查Rx.Observable.prototype.flatMap等的源代码！
你可以使用.mergeAll() ，如下所示：
Rx.Observable.prototype.handleFn = function (fn) { const source = this; return Rx.Observable.create(function (obs) { return source.subscribe(function(val){ obs.next(fn.call(obs,val)); }); }); }; const obs = Rx.Observable.interval(1000) .handleFn(function(){ return Rx.Observable.timer(150).mapTo(Math.random()); }) .mergeAll(); obs.subscribe(x => console.log(x));
在这里看到现场JSBin。
选项2 ：而不是mergeAll ，你也可以做到以下几点：
Rx.Observable.prototype.handleFn = function (fn) { const source = this; return Rx.Observable.create(function (obs) { return source.subscribe(function(val){ fn.call(obs,val).subscribe(x => obs.next(x)); }); }); };
额外的注意事项 ：如果你想看看这是如何正确实现，请看一下（如你已经提到自己）在flatMap, switchMap, concatMap 。
所以看着源代码，它是用TypeScript编写的（对他们来说很好）。 但我不知道TS，所以我把它转换成了ES5。 原来flatMap和mergeMap应该是等价的运算符，而mergeMap是源码中唯一的一个，在这里它被转换：
"use strict"; var __extends = (this && this.__extends) || function (d, b) { for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p]; function __() { this.constructor = d; } d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __()); }; var subscribeToResult_1 = require("../util/subscribeToResult"); var OuterSubscriber_1 = require("../OuterSubscriber"); /* tslint:disable:max-line-length */ /** * Projects each source value to an Observable which is merged in the output * Observable. * * <span class="informal">Maps each value to an Observable, then flattens all of * these inner Observables using {@link mergeAll}.</span> * * <img src="./img/mergeMap.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 Observable, and then merging those resulting Observables and * emitting the results of this merger. * * @example <caption>Map and flatten each letter to an Observable ticking every 1 second</caption> * var letters = Rx.Observable.of('a', 'b', 'c'); * var result = letters.mergeMap(x => * Rx.Observable.interval(1000).map(i => x+i) * ); * result.subscribe(x => console.log(x)); * * // Results in the following: * // a0 * // b0 * // c0 * // a1 * // b1 * // c1 * // continues to list a,b,c with respective ascending integers * * @see {@link concatMap} * @see {@link exhaustMap} * @see {@link merge} * @see {@link mergeAll} * @see {@link mergeMapTo} * @see {@link mergeScan} * @see {@link switchMap} * * @param {function(value: T, ?index: number): Observable} 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 * @param {number} [concurrent=Number.POSITIVE_INFINITY] Maximum number of input * Observables being subscribed to concurrently. * @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 merging the results of the Observables obtained * from this transformation. * @method mergeMap * @owner Observable */ function mergeMap(project, resultSelector, concurrent) { if (concurrent === void 0) { concurrent = Number.POSITIVE_INFINITY; } if (typeof resultSelector === 'number') { concurrent = resultSelector; resultSelector = null; } return this.lift(new MergeMapOperator(project, resultSelector, concurrent)); } exports.mergeMap = mergeMap; var MergeMapOperator = (function () { function MergeMapOperator(project, resultSelector, concurrent) { if (concurrent === void 0) { concurrent = Number.POSITIVE_INFINITY; } this.project = project; this.resultSelector = resultSelector; this.concurrent = concurrent; } MergeMapOperator.prototype.call = function (observer, source) { return source.subscribe(new MergeMapSubscriber(observer, this.project, this.resultSelector, this.concurrent)); }; return MergeMapOperator; }()); exports.MergeMapOperator = MergeMapOperator; /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ var MergeMapSubscriber = (function (_super) { __extends(MergeMapSubscriber, _super); function MergeMapSubscriber(destination, project, resultSelector, concurrent) { if (concurrent === void 0) { concurrent = Number.POSITIVE_INFINITY; } var _this = _super.call(this, destination) || this; _this.project = project; _this.resultSelector = resultSelector; _this.concurrent = concurrent; _this.hasCompleted = false; _this.buffer = []; _this.active = 0; _this.index = 0; return _this; } MergeMapSubscriber.prototype._next = function (value) { if (this.active < this.concurrent) { this._tryNext(value); } else { this.buffer.push(value); } }; MergeMapSubscriber.prototype._tryNext = function (value) { var result; var index = this.index++; try { result = this.project(value, index); } catch (err) { this.destination.error(err); return; } this.active++; this._innerSub(result, value, index); }; MergeMapSubscriber.prototype._innerSub = function (ish, value, index) { this.add(subscribeToResult_1.subscribeToResult(this, ish, value, index)); }; MergeMapSubscriber.prototype._complete = function () { this.hasCompleted = true; if (this.active === 0 && this.buffer.length === 0) { this.destination.complete(); } }; MergeMapSubscriber.prototype.notifyNext = function (outerValue, innerValue, outerIndex, innerIndex, innerSub) { if (this.resultSelector) { this._notifyResultSelector(outerValue, innerValue, outerIndex, innerIndex); } else { this.destination.next(innerValue); } }; MergeMapSubscriber.prototype._notifyResultSelector = function (outerValue, innerValue, outerIndex, innerIndex) { var result; try { result = this.resultSelector(outerValue, innerValue, outerIndex, innerIndex); } catch (err) { this.destination.error(err); return; } this.destination.next(result); }; MergeMapSubscriber.prototype.notifyComplete = function (innerSub) { var 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(); } }; return MergeMapSubscriber; }(OuterSubscriber_1.OuterSubscriber)); exports.MergeMapSubscriber = MergeMapSubscriber; //# sourceMappingURL=mergeMap.js.map