java - RX Java 2，接受要添加的新值的 Observable

Question

我正在创建一个返回observable<Location>的LocationHandler类。我可以发送一个新的位置，订阅者可以获得最后添加的位置和任何后续值。

我已经编写了这个类，它可以工作，但我不知道这是否是正确的方法，因为我添加了一个回调，但我闻起来很糟糕。

感谢您的帮助。

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的建议，我用ReplaySubject修改了它。

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

你可以使用Subject而不是 MessageHandler。主题可以同时充当可观察者和订阅者。您可以在 LocationHandler 中有一个方法，该方法返回您将订阅的 Subject#asObservable。在内部，当setLocation时，您必须调用Subject#onNext来提供位置。有不同类型的主题可供选择。请参阅文档以选择更适合您需求的一种。例如

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

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

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

从外部调用getLocation并订阅返回的Observable。当调用 setLocation 时，您将获得对象 onNext