RxJava - 如何停止(和恢复)一个 Hot Observable(间隔)? [英] RxJava - How to stop (and resume) a Hot Observable (interval)?

问题描述

我有以下 Hot Observable:

I have the following Hot Observable:

hotObservable = Observable.interval(0L, 1L, TimeUnit.SECONDS)
                          .map((t) -> getCurrentTimeInMillis()))

但是，我找不到阻止它的好方法.我能够使用 takeWhile 和一个 boolean 标志(runTimer)部分解决这个问题:

However, I can't find a good way to stop it. I was able to partially solve this using takeWhile and a boolean flag (runTimer):

Observable.interval(0L, 1L, TimeUnit.SECONDS)
          .takeWhile((t) -> runTimer)
          .map((t) -> getCurrentTimeInMillis()))

不过，我不喜欢这种方法的两件事:

There are 2 things I don't like in this approach though:

1. 我必须保留标志 runTimer，这是我不想要的.
2. 一旦 runTimer 变为 false，Observable 就会简单地完成，这意味着如果我想再次发出，我需要创建一个新的 Observable.我不想要那个.我只是想让 Observable 停止发射项目，直到我告诉它重新开始.

1. I must keep the flag runTimer around, which I don't want.
2. Once runTimer becomes false, the Observable simply completes, which means if I want to emit again I need to create a new Observable. I don't want that. I just want the Observable stop emitting items until I tell it to start again.

我希望有这样的事情:

hotObservable.stop();
hotObservable.resume();

这样我就不需要保留任何标志，并且 observable 始终处于活动状态(尽管它可能不会发出事件).

That way I don't need to keep any flags around and the observable is always alive (it might not be emitting events though).

我怎样才能做到这一点?

How can I achieve this?

推荐答案

一种可能的方法是使用 BehaviorSubject 和 switchMap:

One possible approach uses a BehaviorSubject and a switchMap:

BehaviorSubject<Boolean> subject = BehaviorSubject.create(true);
hotObservable = subject.distinctUntilChanged().switchMap((on) -> {
    if (on) {
        return Observable.interval(0L, 1L, TimeUnit.SECONDS);
    } else {
        return Observable.never();
    }
}).map((t) -> getCurrentTimeInMillis());

通过向主题发送布尔值，可以控制 observable 的输出.subject.onNext(true) 将导致使用该主题创建的任何 observable 开始发出值.subject.onNext(false) 禁用该流.

By sending booleans to the subject the output of the observable can be controlled. subject.onNext(true) will cause any observable created using that subject to begin emitting values. subject.onNext(false) disables that flow.

switchMap 在关闭时负责处理底层的 observable.它还使用 distinctUntilChanged 来确保它不会进行不必要的切换.

The switchMap takes care of disposing the underlying observables when it is switched off. It also uses distinctUntilChanged to make sure it does not do unnecessary switching.

这篇关于RxJava - 如何停止(和恢复)一个 Hot Observable(间隔)?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！