My scenario: a controller retrieves channels (/asset/data/media same same) two different components would like to receive the response.
problem: I don't want to run the BE request twice, and I don't want the components "know" that the controller is using rxJava.

Controller:
ct'r
public TvGuideChannelsController(String channelId) {
....
        createChannelObservable();
    }

I created ConnectableObservable so I can replay the result without another server request.
the connect is needed only once.

  protected void createChannelObservable() {

        channelsObservable = Observable.create(new ObservableOnSubscribe<List<AssetInfo>>() {
            @Override
            public void subscribe(@NonNull final ObservableEmitter<List<AssetInfo>> e) throws Exception {

                TvpapiApi.getInstance().requestChannelMediaListUsingGetChannelMultiFilter(mChannelID, 250, 			APIConstants.MediaOrderByEnum.ABC, 0, new OnTvinciAPIResponse<List<AssetInfo>>() {

                    @Override
                    public void onData(TvpapiApi.APIResponse<List<AssetInfo>> data) {
                        mChannels = data.getResponse();
                        EpgManager.getInstance().setMediaChannels(data.getResponse());
                        e.onNext(data.getResponse());
                    }

                    @Override
                    public void onError(TvpapiApi.APIErrorTypes type) {
                        e.onError(new Throwable(type.getExtraReason()));
                    }
                });
            }
        }).replay();

        channelsObservable.connect();
    }

when a component want to use the response it requestChannels with TvGuideResponse interface (not consumer) so the component will not know how the controller's implementation.

@Override
    public void requestChannels(final TvGuideResponse response) {

        subscribeOnChannels(new Consumer<List<AssetInfo>>() {
            @Override
            public void accept(List<AssetInfo> assets) {
                response.onChannels(assets);
            }
        });
    }

the contoller creates a new constumer for each component request and the observable will replay the response for each one.

private void subscribeOnChannels(final Consumer<List<AssetInfo>> assetsConsumer) {
        if (channelsObservable != null){
            channelsObservable.subscribe(new io.reactivex.functions.Consumer<List<AssetInfo>>() {
                @Override
                public void accept(@NonNull List<AssetInfo> assets) throws Exception {
                    assetsConsumer.accept(assets);
                }
            });
        }
    }