Wie kann ich in RxJava Observable aktualisieren?

Question

Ich habe einen beobachtbaren, die eine HTTP-Anfrage

mObservable = retryObservable(mService.getAddressList(getUserId(), true, 1, Integer.MAX_VALUE, "id", true) 
      .map(r -> { 
       return r.getItems(); 
      }) 
      .observeOn(AndroidSchedulers.mainThread())); 

dann Abonnement

mSubscription = mObservable.subscribe(items -> { 
       mAddressAdapter.swapItems(items); 
      }, getActivityBase()::showError); 

hüllt Wenn das Abonnement Initialisierung kommt, kalt beobachtbaren aktiviert und HTTP-Anforderung ausgelöst wird. Jetzt weiß ich, dass sich die zugrunde liegenden Daten geändert haben, und ich muss dieselbe, aber neue Anfrage machen. Ich habe

versucht

mSubscription.unsubscribe(); 

dann

Aufruf

mObservable.subscribe(items -> {doSomething();}) 

wieder, wie aus meinem Verständnis sollte die beobachtbare auslösen abonnieren möchte, aber es funktioniert nicht. Irgendwelche Vorschläge?

Answer 1

Sobald die Observable abgeschlossen ist, veröffentlicht es keine neuen Elemente. Es ist Rx Vertrag.

Fügen Sie Ihren Code in eine Methode ein und erstellen Sie jedes Mal eine neue Observable.

Observable<?> getObservable() { 
    return retryObservable(mService.getAddressList(getUserId(), true, 1, Integer.MAX_VALUE, "id", true) 
      .map(r -> { 
       return r.getItems(); 
      }) 
      .observeOn(AndroidSchedulers.mainThread())); 
} 

Wie @DaveSexton im Kommentar erwähnt gibt es noch besseren Ansatz defer Funktion in RxJava

erstellen Sie nicht die beobachtbare mit, bis ein Teilnehmer abonniert haben; eine frischen Observable auf jedes Abonnement erstellen

Pass DEFER() eine beobachtbare Werksfunktion (eine Funktion, dass erzeugt Observablen) und aufzuschieben() gibt einen beobachtbaren zurückkehren dass diese Funktion aufrufen wird seine Observable Sequenz von neuem zu erzeugen jedes Mal, wenn ein neuer Abonnent abonniert wird.

Mehr hier: https://github.com/ReactiveX/RxJava/wiki/Creating-Observables#defer