RxJava caching

gistfile1.java

private PublishSubject<Data> dataRequest;
private Data data;

// ...

Subscription getData(Observer<Data> observer) {
    if (data != null) {
        observer.onNext(data);
    }

    if (dataRequest != null) {
        return dataRequest.subscribe(observer);
    }

    dataRequest = PublishSubject.create();

    Subscription subscription = dataRequest.subscribe(observer);

    dataRequest.subscribe(new EndObserver<Data>() {
        @Override
        public void onEnd() {
            dataRequest = null;
        }

        @Override
        public void onNext(Data data1) {
            data = data1;
        }
    });

    apiService.data()
            .map(this::cache)
            .subscribeOn(ioScheduler)
            .observeOn(uiScheduler)
            .subscribe(dataRequest);

    return subscription;
}