检查是否未完成 Observable 是否为空答案

【问题标题】：check if not completed Observable is empty检查是否未完成 Observable 是否为空
【发布时间】：2017-05-25 22:54:05
【问题描述】：

有没有一种好方法可以检查未完成的 Observable 在那个确切时间是否为空？

let cache = new ReplaySubject<number>(1);
...
// Here I want to know if 'cache' still empty or not. And, for example, fill it with initial value.
cache.isEmpty().subscribe(isEmpty => {
    if (isEmpty) {
        console.log("I want to be here!!!");
        cache.next(0);
    }
});
// but that code does not work until cache.complete()

【问题讨论】：

“空”Observable 在这种情况下并不是一个真正有意义的概念，我认为您将Observer 和Observable 之间的问题混为一谈。你想通过这个来完成什么？

标签： javascript typescript rxjs rxjs5

【解决方案1】：

你可以使用takeUntil():

Observable.of(true)
    .takeUntil(cache)
    .do(isEmpty => {
        if (isEmpty) {
            console.log("I want to be here!!!");
            cache.next(0);
        }
    })
    .subscribe();

但是这只会工作一次。

另一种方法是将缓存“清空”并使用BehaviorSubject 将其初始化为空：

let cache = new BehaviorSubject<number>(null as any);
...
cache
   .do(content => {
       if (content == null) {
           console.log("I want to be here!!!");
           cache.next(0);
       }
    })
    .subscribe();

当然，您可以立即使用一些默认值初始化缓存。

【讨论】：

此外，BehaviorSubject 有一个方法 getValue 可以返回当前值（如果有）。

【解决方案2】：

实际上，这并不是那么简单，而且公认的答案也不是很普遍。您想检查 ReplaySubject 在此特定时间点是否为空。

但是，如果您想让它真正与 ReplaySubject 兼容您还需要考虑 windowTime 参数，该参数指定 “生存时间”通过此对象的每个值。这意味着您的cache 是否为空会随时间变化。

ReplaySubject 有方法 _trimBufferThenGetEvents 可以满足您的需求。不幸的是，这个方法是私有的，所以你需要在 JavaScript 中做一点“hack”并直接扩展它的prototype。

import { ReplaySubject } from 'rxjs';

// Tell the compiler there's a isNowEmpty() method
declare module "rxjs/ReplaySubject" {
    interface ReplaySubject<T> {
        isNowEmpty(): boolean;
    }
}

ReplaySubject.prototype['isNowEmpty'] = function() {
    let events = this._trimBufferThenGetEvents();
    return events.length > 0;
};

那么使用这个ReplaySubject就很简单了：

let s = new ReplaySubject<number>(1, 100);
s.next(3);
console.log(s.isNowEmpty());
s.next(4);

setTimeout(() => {
    s.next(5);
    s.subscribe(val => console.log('cached:', val));
    console.log(s.isNowEmpty());
}, 200);

setTimeout(() => {
    console.log(s.isNowEmpty());
}, 400);

请注意，一些对isNowEmpty() 的调用返回true，而另一些则返回false。例如最后一个返回false，因为该值在此期间无效。

这个例子打印：

true
cached: 5
true
false

观看现场演示：https://jsbin.com/sutaka/3/edit?js,console

【讨论】：

我也在查看_trimBufferThenGetEvents() 函数来实现这一点，这似乎是唯一的方法。可惜没有暴露这个功能。

【解决方案3】：

您可以使用.scan() 来累积您的计数，并将其映射到一个布尔值（无论它是否为非零）。（它需要第二个参数作为种子值，使其以 0 开头，因此它始终反映当前计数。）

我还添加了一个 .filter() 而不是 if 语句以使其更简洁：

let cache = new ReplaySubject<number>(1);

cache
    .map((object: T) => 1)
    .scan((count: number, incoming: number) => count + incoming, 0)
    .map((sum) => sum == 0)
    .filter((isEmpty: boolean) => isEmpty)
    .subscribe((isEmpty: boolean) => {
        console.log("I want to be here!!!");
        cache.next(0);
    });

【讨论】：

【解决方案4】：

startWith

let cache = new ReplaySubject<number>(1);

isEmpty$ = cache.pipe(mapTo(false), startWith(true));

这说：

无论缓存发出什么值 - 将其映射到 false。（因为发射后它不是空的）
如果尚未发出任何内容，则以 true 开头（因为这意味着它是空的）

【讨论】：