RxJs：在 Observable<T> 数组上的“映射”，但只要 T 匹配条件就返回答案

【问题标题】：RxJs : 'Map' on an Array of Observable<T>, but returning as soon as a T match a conditionRxJs：在 Observable<T> 数组上的“映射”，但只要 T 匹配条件就返回
【发布时间】：2016-07-23 13:36:22
【问题描述】：

上下文：使用 Angular2 在 TypeScript 中 + rxjs 5 编写的应用程序。

编辑：我准确地说我对 Rx 库和事情应该以“惯用方式”完成的方式相对较新。是的，在发布到 SO 之前，我试图在文档中找到一些线索。

我有这个：

class Result { constructor(public inError: boolean) { } }

const checks : Array<() => Observable<Result>> = [...];

这是一个函数数组，每个函数返回一个包含 Result 对象的 observable。

我想要什么：

我想将此数组“映射”到Array<Observable<Result>>，基本上是通过依次调用每个函数...
...但我想在第一个Result.inError 为真后立即打破“地图”！

我绝望地陷入困境，摆弄reduce、takeWith、contains等... Observable 的延迟特性让我感到困惑。

任何帮助将不胜感激！

【问题讨论】：

也许投反对票的人可以解释为什么他对我的问题投反对票？

标签： typescript angular rxjs

【解决方案1】：

如果您的 observable 正在执行同步操作，您可以简单地这样做：

const results = [];
for (const i = 0; i < checks.length; i +=1) {
    checks[i]().subscribe(result => if (result.inError) {
        break;
    } else {
        results.push(checks[i]());
    });
}
// Results observables accessible here.

在异步观察者的情况下：

const results = [];
function callObservable(index) {
    checks[index]().subscribe(result => if (result.inError) {
            // Results observables accessible here.
        } else {
            results.push(checks[i]());
            callObservable(index + 1);
        })
}
callObservable(0);

不过，这样做并没有带来任何好处。你的 observables 在到达结果数组之前就已经被调用了，或者如果再次从这个数组调用的话，将会有另一个值。

【讨论】：

【解决方案2】：

经过越来越多的挖掘，这里有一个解决方案：

// Some class that will be contained in my observables
class Result { constructor(public inError: boolean) { } }

// An array of Observables that will emit those classes instances
// the thunk is just here to lazy instantiate the Result classes
// only when needed
const oResults : Array<() => Observable<Result>> = [
  Rx.Observable.of(() => new Result(false)),
  Rx.Observable.of(() => new Result(false)),
  Rx.Observable.of(() => new Result(true)),
  Rx.Observable.of(() => new Result(false))
];

// An home made (found on SO) INCLUSIVE 'takeWhile' version
const takeWhileInclusive(source, predicate){
    return source.publish(co => co.takeWhile(predicate)
        .merge(co.skipWhile(predicate).take(1)));
}

// Let's filter out things as expected
const res = takeWhileInclusive(
    Rx.Observable.merge(oResults).map( x => x()), 
    x => !x.inError
);

// I can now subscribe to the resulting stream and will only get
// all the first Results that are false AND the first following result that
// is true
res.subscribe(next => console.info("Next result", next));

【讨论】：