Опрос сервера с использованием RxJava - повторить после получения ответа сервера

Question

В настоящее время я пытаюсь реализовать опрос серверов с использованием RxJava, я провел некоторое исследование о том, как повторить всю цепочку после получения ответа сервера, я попытался использовать repeat(), it работает, но не работает, что идеально, причина в том, что он делает api call столько раз, и серверу требуется дополнительное время для обработки данных перед отправкой клиенту, но мы не знаем точное время, поэтому мы не можем использовать repeatWhen(), чтобы дать конкретное время. Единственное, что я могу использовать, - это подождать после ответа api.

Любой совет будет оценен!

Ниже фрагмент кода:

retrofitService.requestPolling() 
       .repeat() // do not wait to call server so many times 
       .takeUntil(new Func1<PollResponse, Boolean>() { 
         @Override 
         public Boolean call(PollResponse pollResponse) { 
          return pollResponse.mComplete; 
         } 
       }) 
      .doOnNext(new Action1<FlightSearchPollResponse>() { 
       @Override 
       public void call(pollResponse pollResponse) { 
         // update UI here 
       } 
      }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Observer<PollResponse>() { 
       @Override 
       public void onCompleted() { 

        } 
       } 

       @Override 
       public void onError(Throwable e) { 

       } 

       @Override 
       public void onNext(PollResponse pollResponse) { 

      }); 

EDIT: Я новичок в RxJava, только что получил эту тему под названием противодавление, и есть много статей объяснили, как бороться с этим, так как я не хочу кешируйте этот ответ, кажется, что Тема будет хорошим вариантом, это позволяет вам контролировать, когда тянуть.

http://akarnokd.blogspot.com/2015/06/subjects-part-1.html

И благодаря @Gary LO

Answer 1

Там должно быть много подходов. Я хотел бы поделиться одним из них.

1. Создание отдельного потока сигналов PublishSubject pollingSignal
2. преобразование сигнала к апи вызова

3. Опубликовать сигнал, чтобы сделать это снова.

   final PublishSubject<Boolean> pollingSignal = PublishSubject.create(); 
   
   final Observable<PollResponse> apiResponse = retrofitService.requestPolling(); 
   
   pollingSignal 
       .flatMap(x -> apiResponse) 
       .subscribe(new Observer<PollResponse>() { 
       @Override 
       public void onCompleted() {} 
   
       @Override 
       public void onError(Throwable throwable) {} 
   
       @Override 
       public void onNext(PollResponse integer) { 
        // start the next polling 
        pollingSignal.onNext(true); 
       } 
   }); 
   
   // start the first polling 
   pollingSignal.onNext(true); 
   

Удачи!

Примечания Использование PublishSubject<Boolean> вместо PublishSubject<Void> потому, что я не чувствую себя комфортно использовать pollingSignal.onNext(null).

Но в Котлин я могу использовать PublishSubject<Unit> с pollingSignal.onNext(Unit)