Rx 超时运算符

Question

我想在 x 秒后发射一个物品，如果没有物品被发射的话。我正在尝试使用超时运算符。问题是超时运算符需要至少执行一项才能开始倒计时。来自文档：

"If the next item isn't emitted within the specified timeout duration starting from its predecessor, the resulting Observable begins instead to mirror a fallback Observable."

这不是我想要的行为。当我订阅我的 observable 时，我想在特定时间段过去而之前没有任何发射项目的情况下发射一个特定项目。

示例：

 getUserLocationFromGPS() //Sometimes I dont receive user location
.timeout(5, TimeUnit.SECONDS, Observable.just(getDefaultLocation())
.subscribe(...);

Answer 1

final Observable data = yourApi.getData().take(5, TimeUnit.SECONDS);
final Observable fallback = Observable.just("Fallback");

Observable.concat(data, fallback).firstOrError()
            .subscribe(
                    System.out::println
            );

如果 yourApi.getData() 没有发出任何东西，那么您将从您的回退可观察对象中接收数据。
如果 yourApi.getData() 超时，那么您将从后备可观察对象中接收数据。
如果yourApi.getData()正常发射，那么您将收到来自它的数据。

Answer 2

我没有具体使用 java-rx，但这里有一个想法：

getUserLocationFromGPS()
.buffer(5 sec, 1 item) //not sure how to do this in rx-java
.map(x => { // this is select in .net
    if(x.IsEmpty)
      return getDefautLocation();
    else
      return x[0];
    }).Subscribe(...)

buffer 将每 5 秒为您获取项目列表。您可以在 getUserLocationFromGPS()

仅发出 1 个项目后将其指定为 return

map 将检查列表是否为空（未发出任何项目）和 return 正确值

Rx 超时运算符

Rx Timeout Operator

rx-java

rx-android