RxStore

RxStore.java

public abstract class RxStore<Input, Output> {
    private BehaviorSubject<Input> subject = BehaviorSubject.create();
    private Observable<Output> observable;

    public final Subscription register(Subscriber<Output> subscriber) {
        if (observable == null) {
            observable = subject.compose(getTransformer());
        }

        return observable.subscribe(subscriber);
    }

    public final void unregister(Subscription subscription) {
        if (subscription != null && !subscription.isUnsubscribed()) {
            subscription.unsubscribe();
        }
    }

    protected abstract Observable.Transformer<Input, Output> getTransformer();

    public final void onNext(Input event) {
        subject.onNext(event);
    }
}