仅在链中的 Observable 发出事件后才获取事件

Question

这里是链：observableOne.flatMap(event -> observableTwo).subscribe()

我想在observableOne 第一次发出一个事件时从它获取事件，然后忽略来自这个Observable 的所有其他事件，直到observableTwo 发出一个值、完成或完成一个错误之后。值得一提的是，我最终对第二个 Observable 中的事件感兴趣。

上下文就像，有一个Button点击触发一个事件，这是ObservableOne。触发的事件触发ObservableTwo，假设它是一个网络操作链。所以我想在执行网络操作时忽略所有按钮点击。

Answer 1

您可以使用Flowable 而不是Observable，然后您可以通过将flatMap 的maxConcurrency 设置为1 来使用背压来实现所需的效果（删除事件直到observableTwo 终止） ：

observableOne
    .toFlowable(BackpressureStrategy.DROP)
    .flatMap(event -> observableTwo.toFlowable(BackpressureStrategy.ERROR), 1)
    .subscribe();

我发布了similar question here。

原来在 akarnokd 的 RxJava2Extensions 库中也有一个专门用于此目的的 ObservableTransformer。可以这样使用：

observableOne
    .compose(ObservableTransformers.flatMapDrop(event -> observableTwo))
    .subscribe();

Answer 2

要控制flatMap 的请求数量，请使用特殊的重载：

observableOne
  .doOnRequest(System.out::println)
  .flatMap(event -> observableTwo, 1)
  .flatMap(event -> observableThree, 1)
.subscribe()

如果您的源 observableOne、observableTwo、observableThree 是同步的，则这不是必需的，但对于异步源，这应该可以完成工作。

Answer 3

[Edit2] 问题发生了变化，所以我调整了我的答案 + 实际上给出了正确答案

除了使用状态标志之外，我没有看到其他方法：

AtomicBoolean progress = new AtomicBoolean();
observableOne
        .filter(event -> !progress.get())
        .flatMap(event ->
                observableTwo
                        .doOnSubscribe(() -> progress.set(true))
                        .doOnTerminate(() -> progress.set(false))
        )
        .subscribe();

如果发生错误，您的订阅将被取消，并且您将不会再收到任何事件，即使再次单击该按钮也是如此。

如果这不是你想要的，你可以：

• 重新订阅错误回调

  private void bindRemoteCalls() {
      if (mySubscription != null) mySubscription.unsubscribe();
      AtomicBoolean progress = new AtomicBoolean();
      mySubscription = observableOne
          .filter(event -> !progress.get())
          .flatMap(event ->
                  observableTwo
                          .doOnSubscribe(() -> progress.set(true))
                          .doOnTerminate(() -> progress.set(false))
          )
          .flatMap(event -> observableTwo, 1)
          .subscribe(
              data -> handleResponse(data),
              error -> {
                  handleError(error);
                  bindRemoteCalls();
              }
          );
  }
  

• 使用onErrorReturn()（结合doOnError() 来实际做一些事情）

  AtomicBoolean progress = new AtomicBoolean();
  observableOne
      .filter(event -> !progress.get())
      .flatMap(event ->
              observableTwo
                      .doOnSubscribe(() -> progress.set(true))
                      .doOnTerminate(() -> progress.set(false))
                      .doOnError(error -> handleError(error))
                      .onErrorReturn(null)
                      .filter(data -> data != null)
      )
      .subscribe(data -> handleResponse(data));
  

如果需要，请记住将subscribeOn() / observeOn() 与正确的调度程序一起使用。

请考虑使用switchMap() 代替flatMap() --> 如果再次按下按钮，则先前的呼叫被取消（取消订阅），然后开始新的呼叫。或者用 Rx 术语来说：observableTwo 之前的订阅被取消订阅，并形成了一个新的订阅。

如果您在取消订阅时禁用按钮并在终止时启用它，您可以通过使按钮不可点击来轻松获得相同的结果。