rxjs - 从不完整的 observable 中获取最后一个值

Question

有一个不完整的 observable 可以有或没有 n 值的重放。我想从中获取最后一个值 - 如果还没有，则只获取下一个值。

这适用于 first() 的第一个可用值和 take(1) ( example ):

possiblyReplayedIncomplteObservable.first().toPromise().then(val => ...);

但是对于最后一个值 last()和 takeLast(1)等待可观察的完成 - 这里不是理想的行为。

如何解决这个问题？是否有特定的运营商？

Answer 1

我有 ReplaySubject(2) 的解决方案“耗尽”序列以获取最新元素，如果序列为空，则仅获取最后一个元素，但是，它很麻烦并且不能很好地扩展(例如，如果您决定将重放大小增加到 3)。然后我想起重放/行为主题在管道传输时往往难以管理。最简单的解决方案是创建一个“阴影”序列并将您的 ReplaySubject 通过管道传输到其中(而不是通过对您的 ReplaySubject 进行转换/操作来创建它)，因此:

var subject$ = new Rx.ReplaySubject(3);
var lastValue$ = new Rx.ReplaySubject(1);
subject$.subscribe(lastValue$); // short hand for subject$.subscribe(v => lastValue$.next(v))

lastValue$.take(1).toPromise().then(...);

========== 旧解决方案，忽略 ReplaySubject(2) ==================

阅读下面的评论后，正确的代码是:

Rx.Observable.combineLatest(possiblyReplayedIncomplteObservable).take(1).subscribe(...)

并不是

Rx.Observable.combineLatest(possiblyReplayedIncomplteObservable).subscribe(...)

这是因为 promise 是“一次性”可观察的。我认为 toPromise()代码仅在完成时解析结果。
take(1)不会影响您的原始流，因为它对由 combineLatest 创建的新流进行操作。 .

实际上，最简单的方法是:

possiblyReplayedIncomplteObservable.take(1).toPromise().then(...)