Возьмите события только после того, как Наблюдаемая из цепи испустила один

Question

Вот цепочка: observableOne.flatMap(event -> observableTwo).subscribe()

Я хотел бы взять события из observableOne самых первый раз, когда он испускает один, а затем игнорировать все другие события из этого Observable до тех пор, пока observableTwo не испустит значение, не завершит или не завершит ошибку. Стоит упомянуть, что в конечном итоге меня интересуют события со второго Наблюдаемого.

Контекст такой, как есть, есть Button, нажимая на которое происходит событие, это ObservableOne. Вызываемое событие запускает ObservableTwo, скажем, это цепочка сетевых операций. Поэтому я хочу игнорировать все нажатия кнопок при выполнении сетевых операций.

Answer 1

Для управления запросами на суммы, сделанные flatMap использовать специальную перегрузку:

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

Если ваши источники observableOne, observableTwo, observableThree синхронны это не должно быть необходимым, но для асинхронных источников, это должно сделать работу.

Answer 2

[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: наблюдаемая две предыдущие подписки отписаны и сформирована новая подписка.

Если вы отключите свою кнопку при отмене подписки и включите ее при завершении, вы можете легко получить тот же результат, сделав кнопку незаметной.