Rx 超时运算符

Question

如果没有发出任何项目，我想在 x 秒后发出一个项目。我试图使用超时运算符。问题是超时运算符需要至少执行一项才能开始倒计时。来自文档：

“如果下一个项目没有在从其前身开始的指定超时时间内发出，则生成的 Observable 开始镜像一个备用 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
            );

1. 如果 yourApi.getData() 没有发出任何内容，那么您将从后备 observable 接收数据。

2. 如果 yourApi.getData() 超时，那么您将从后备 observable 接收数据。

3. 如果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 个项目后返回

map 将检查列表是否为空（没有发出任何项目）并返回正确的值