RxJava Polling + manuelle Aktualisierung

Question

Ich habe eine Liste, die jede Minute aktualisiert werden soll. Zum Beispiel die Benutzerliste hier: https://github.com/android10/Android-CleanArchitecture/blob/master/domain/src/main/java/com/fernandocejas/android10/sample/domain/interactor/GetUserList.java

füge ich eine periodische Auffrischung mit repeatWhen:

public Observable<List<User>> buildUseCaseObservable(Void unused) { 
    return this.userRepository 
     .users() 
     .repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() { 
      @Override 
      public ObservableSource<?> apply(Observable<Object> objectObservable) throws Exception { 
      return objectObservable.delay(1, TimeUnit.MINUTES); 
      } 
     }); 
    } 

Es funktioniert auf diese Weise, jede Minute OnNext aufrufen. Aber wenn ich diese Liste sofort aktualisieren möchte (wegen der Aktion des Benutzers oder wegen einer Benachrichtigung), weiß ich nicht, wie ich das ausführen soll.

Sollte ich das Observable abbrechen/entsorgen und einen neuen starten? Dank

Answer 1

Aus Ihrem Code verstehe ich, dass die Benutzerliste generiert und beim Abonnement ausgegeben wird.

Hier sind einige Lösungen, die ich denken kann, statt abzumelden und resubscribing auf das Ereignis, zu dem Sie sofort reagieren wollen:

1. Statt den repeatWhen Betreiber, verwenden Sie die interval creation operator mit dem flatMap kombiniert um jede Minute das Abonnement für ein neues Observable aufzurufen und den Operator merge zu verwenden, um die Reaktion auf das andere Ereignis, an dem Sie interessiert sind, hinzuzufügen. Etwas wie folgt aus:

   @Test 
   public void intervalObservableAndImmediateReaction() throws InterruptedException { 
       Observable<String> obs = Observable.interval(1, TimeUnit.SECONDS) 
          .cast(Object.class)           
          .mergeWith(
             Observable.just("mockedUserClick") 
               .delay(500, TimeUnit.MILLISECONDS)) 
          .flatMap(
            timeOrClick -> Observable.just("Generated upon subscription") 
            ); 
   
       obs.subscribe(System.out::println); 
       Thread.currentThread().sleep(3000); //to see the prints before ending the test 
   } 
   

   oder angepasst an Ihre Bedürfnisse (aber das Prinzip ist das gleiche):

   Observable.interval(1, TimeUnit.MINUTES) 
         .mergeWith(RxView.clicks(buttonView)) 
         .flatMap(timeOrClick -> this.userRepository.users()); 
   

2. Sie den flatMap Betreiber nach wie vor verwenden können, auch wenn Sie halten aktuelle Implementierung arbeiten und ohne auf ein Intervall zu verschmelzen - nur halten Ihre Arbeits Code und in einem anderen Bereich der Programmkette an die RxBinding Ihre Wahl:

   RxView.touches(yourViewVariable) 
        .flatMatp(motionEvent -> this.userRepository.users()) 
        .subscribe(theObserver); 
   

   Beachten Sie, dass das Abonnement in dieser Lösung unabhängig von den beiden Observablen erfolgt. Sie werden wahrscheinlich besser dran sein, wenn Sie verschiedene Beobachter verwenden oder ein Thema oder etwas in dieser Zeile verwalten. Ein kleiner Test, den ich ausführte, zeigte, dass ein Abonnent ohne Probleme zwei verschiedene Observables abonniert hat (in Rxjava1 - habe Rxjava2 noch nicht eingecheckt), aber es fühlt sich für mich zweifelhaft an.

Answer 2

Wenn Sie nicht betroffen sind mit der Refresh-Zeit nach einer der anderen Observablen Einstellung sendet Daten, die Sie tun können, etwa wie folgt:

// Specific example of a user manually requesting 
    val request = Observable.create<String> { emitter -> 
     refresh.setOnClickListener { 
      emitter.onNext("Click Request") 
     } 
    } 
      .observeOn(Schedulers.io()) 
      .flatMap { 
       userRepository.users() 
      } 

    // Refresh based off of your original work, could use something like interval as well 
    val interval = userRepository.users() 
      .subscribeOn(Schedulers.io()) 
      .repeatWhen { objectObservable -> 
       objectObservable.delay(1, TimeUnit.MINUTES) 
      } 

    // Combine them so that both emissions are received you can even add on another source 
    Observable.merge(request,interval) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe({ 
       contents.text = it.toString() 
      }, { 
       contents.text = it.toString() 
      },{ 
       println(contents.text) 
      }) 

Dann müssen Sie nicht entsorgen und resubscribe jedes Mal