获取 Observable 的最新值并立即发出

Question

我正在尝试获取给定的最新值 Observable 并在调用后立即发出。以下面的代码为例：

 return Observable.just(myObservable.last())
    .flatMap(myObservable1 -> {
        return myObservable1;
    })
    .map(o -> o.x) // Here I want to end up with a T object instead of Observable<T> object

这是行不通的，因为这样做 flatMap 将发射 myObservable1 反过来又必须发射才能到达 map 我不知道是否有可能做这样的事情。有没有人知道如何实现这个目标？谢谢

原文由 E-Kami 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

last() 方法在这里没有任何帮助，因为它等待 Observable 终止给你最后一个发出的项目。

假设您无法控制发射的可观察对象，您可以简单地创建一个 BehaviorSubject 并将其订阅到发出您想要收听的数据的可观察对象，然后订阅创建的主题。由于 Subject 既是 Observable 又是 Subscriber 你会得到你想要的。

我认为（现在没有时间检查）您可能必须手动取消订阅原始可观察对象，因为 BehaviorSubject 一旦他的所有订阅者取消订阅将不会自动取消订阅。

是这样的：

 BehaviorSubject subject = new BehaviorSubject();
hotObservable.subscribe(subject);
subject.subscribe(thing -> {
    // Here just after subscribing
    // you will receive the last emitted item, if there was any.
    // You can also always supply the first item to the behavior subject
});

http://reactivex.io/RxJava/javadoc/rx/subjects/BehaviorSubject.html

原文由 MatBos 发布，翻译遵循 CC BY-SA 3.0 许可协议

Answer 2

在 RxJava 中，subscriber.onXXX 被称为异步。这意味着如果您的 Observable 在新线程中发射项目，则您永远无法在返回之前获得最后一个项目，除非您阻塞线程并等待项目。但是如果 Observable 同步发射项目而且你不通过subscribeOn和observOn改变它的线程，比如代码：

 Observable.just(1,2,3).subscribe();

在这种情况下，您可以通过这样做来获取最后一项：

 Integer getLast(Observable<Integer> o){
    final int[] ret = new int[1];
    Observable.last().subscribe(i -> ret[0] = i);
    return ret[0];
}

这样做是个坏主意。RxJava 更喜欢你用它来做异步工作。

原文由 dieyidezui 发布，翻译遵循 CC BY-SA 3.0 许可协议