从另一个可观察到的无限可观察

Question

我有一个 Observable 表示从 DB table 中选择的序列，所以它是有限的。

Observable<Item> selectResults() { ... }

我想实现一个指定间隔的拉动，所以最后我会得到另一个 observable，它将包裹我原来的并无限期地拉动。

我只是不知道该怎么做:(

好的，这是我的想法，围绕可观察的间隔建模，可能需要错误处理和取消订阅逻辑。

public class OnSubscribePeriodicObservable implements OnSubscribe<Item> {
...

  @Override
  public void call(final Subscriber<? super Item> subscriber) {
      final Worker worker = scheduler.createWorker();
      subscriber.add( worker );

      worker.schedulePeriodically(new Action0() {
          @Override
          public void call() {
    selectResults().subscribe( new Observer<Item>() {
            @Override
            public void onCompleted() {
              //continue
            }

            @Override
            public void onError(Throwable e) {
              subscriber.onError( e );
            }

            @Override
            public void onNext(Item t) {
              subscriber.onNext( t );
            }
         }); 
          }

      }, initialDelay, period, unit);
}

Answer 1

您可以使用标准运算符完成此操作，这将为您提供便宜的错误传播、取消订阅和背压：

Observable<Integer> databaseQuery = Observable
    .just(1, 2, 3, 4)
    .delay(500, TimeUnit.MILLISECONDS);

Observable<Integer> result = Observable
        .timer(1, 2, TimeUnit.SECONDS)
        .onBackpressureDrop()
        .concatMap(t -> databaseQuery);

result.subscribe(System.out::println);

Thread.sleep(10000);

从另一个可观察到的无限可观察

Infinite observable from another observable

java

rx-java