RX Java 2, Observable akzeptiert neue Werte zum Hinzufügen

Question

Ich bin auf der Suche nach einer LocationHandler-Klasse, die eine observable<Location> zurückgibt, ich kann einen neuen Standort senden und Abonnenten erhalten die letzte hinzugefügt und alle nachfolgenden Werte.

Ich habe diese Klasse geschrieben, es funktioniert, aber ich weiß nicht, ob es der richtige Weg ist, es zu tun, weil ich einen Rückruf hinzugefügt habe und ich rieche es schlecht.

Danke für jede Hilfe.

public class LocationHandler { 
    private MessageHandler<Location> onNewItem; 
    private Observable<Location> locationObservable; 

    public LocationHandler(LocationInitializationBuilder locationInitBuilder) { 
     locationObservable = getHookedObservable() 
       .mergeWith(locationInitBuilder.build()) 
       .replay(1).autoConnect(); 
    } 


    private Observable<Location> getHookedObservable() { 
     return Observable.create(new ObservableOnSubscribe<Location>() { 
      @Override 
      public void subscribe(ObservableEmitter<Location> e) throws Exception { 
       onNewItem = location -> e.onNext(location); 
      } 
     }); 
    } 

    public Observable<Location> getLocation(){ 
     return locationObservable; 
    } 

    public void setLocation(Location address){ // <---------- add new values 
     if (onNewItem != null){ 
      onNewItem.handleMessage(address); 
     } else { 
      throw new IllegalStateException("Cannot add an item to a never subscribed stream"); 
     } 
    } 
} 

---

@Blackbelt Beratung Nach ich es mit einem ReplaySubject geändert haben.

public class LocationHandler { 
    private ReplaySubject<Location> inputStream = ReplaySubject.create(1); 
    private Observable<Location> locationObservable; 

    public LocationHandler(LocationInitializationBuilder locationInitBuilder) { 
     locationObservable = locationInitBuilder.build() 
       .mergeWith(inputStream) 
       .replay(1).autoConnect(); 
    } 

    public Observable<Location> getLocation(){ 
     return locationObservable; 
    } 

    public void setLocation(Location address){ 
     inputStream.onNext(address); 
    } 
} 

Answer 1

könnten Sie eine Subject statt MessageHandler verwenden. Das Subjekt kann gleichzeitig als beobachtbar und Abonnent fungieren. Sie könnten eine Methode in Ihrer LocationHandler haben, die Subject#asObservable zurückgibt, zu der Sie abonnieren werden. Intern, wenn setLocation, müssen Sie Subject#onNext aufrufen Bereitstellung des Standortes. Es gibt verschiedene Arten von Fächern. Bitte lesen Sie die Dokumentation, um diejenige zu wählen, die Ihren Bedürfnissen besser entspricht. Z.B.

public class LocationHandler { 
    BehaviorSubject<GeevLocation> mLocationSubject = BehaviorSubject.create(); 

    public Observable<GeevLocation> getLocation() { 
     return mLocationSubject.asObservable(); 
    } 

    public void setLocation(GeevLocation address){ 
     mLocationSubject.onNext(address); 
    } 
} 

von außen Anruf getLocation und abonnieren Sie den zurück Observable. Wenn ein setLocation aufgerufen wird, erhalten Sie das Objekt onNext

Answer 2

wie Blackbelt bereits sagte Ihnen, Sie würden ein Thema verwenden. Insbesondere würde ich ein BehaviorSubject verwenden. Betreffs sind standardmäßig heiß, aber sie können Ereignisse nach Abonnement wiedergeben. BehaviorSubject gibt Ihnen den letzten emittierten Wert oder den init-Wert, wenn Sie abonnieren. Jeder Abonnent erhält die Werte als die kommen. Der Stream wird nie beendet, weil es heiß ist. Bitte denken Sie daran, Fehler zu beheben, weil der zweite onError verschluckt wird.

Beispiel Code

class Location { 

} 

class LocationInitializationBuilder { 
    static Location build() { 
     return new Location(); 
    } 
} 

class LocationHandler { 
    private Subject<Location> locationObservable; 

    public LocationHandler(LocationInitializationBuilder locationInitBuilder) { 
     Location initialValue = LocationInitializationBuilder.build(); 

     locationObservable = BehaviorSubject.<Location>createDefault(initialValue).toSerialized(); 
    } 

    public Observable<Location> getLocation() { 
     return locationObservable.hide(); 
    } 

    public void setLocation(Location address) { // <---------- add new values 
     locationObservable.onNext(address); 
    } 
} 

public class LocationTest { 
    @Test 
    public void name() throws Exception { 
     LocationHandler locationHandler = new LocationHandler(new LocationInitializationBuilder()); 

     TestObserver<Location> test = locationHandler.getLocation().test(); 

     locationHandler.setLocation(new Location()); 

     test.assertValueCount(2); 
    } 
}