RxUtil.kt

RxUtil.java

public class RxUtil {

    public static final String LOADING_MESSAGE = "Loading";

    public static <T> ObservableTransformer<T, T> applyUIDefaults(RxFragment rxFragment) {
        return upstream -> upstream
                .compose(RxUtil.addToCompositeDisposable(rxFragment))
                .compose(RxUtil.applySchedulers())
                .compose(RxUtil.applyRequestStatus(rxFragment))
                .compose(RxUtil.showLoadingDialog(rxFragment));
    }

    private static final ObservableTransformer schedulersTransformer =
            observable -> observable.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread());

    private static <T> ObservableTransformer<T, T> applySchedulers() {
        return (ObservableTransformer<T, T>) schedulersTransformer;
    }

    private static <T> ObservableTransformer<T, T> addToCompositeDisposable(RxFragment rxFragment) {
        return upstream -> upstream.doOnSubscribe(disposable -> rxFragment.getCompositeDisposable().add(disposable));
    }

    private static <T> ObservableTransformer<T, T> applyRequestStatus(RxFragment rxFragment) {
        return upstream -> upstream.doOnSubscribe(disposable -> rxFragment.setRequestInProgress(true))
                .doOnTerminate(() -> rxFragment.setRequestInProgress(false));
    }

    private static <T> ObservableTransformer<T, T> showLoadingDialog(RxFragment rxFragment) {
        return observable -> observable
                .doOnSubscribe(disposable -> DialogUtils.showProgressDialog(rxFragment.getFragmentManager(), LOADING_MESSAGE))
                .doOnTerminate(() -> DialogUtils.hideProgressDialog(rxFragment.getFragmentManager()));
    }
}

RxUtil.kt

val LOADING_MESSAGE = "Loading"

fun <T> Observable<T>.applyUIDefaults(rxFragment: RxFragment): Observable<T> =
        this.addToCompositeDisposable(rxFragment)
                .applySchedulers()
                .applyRequestStatus(rxFragment)
                .showLoadingDialog(rxFragment)

private fun <T> Observable<T>.applySchedulers(): Observable<T> = this.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())

private fun <T> Observable<T>.addToCompositeDisposable(rxFragment: RxFragment): Observable<T> =
        this.doOnSubscribe {
            rxFragment.compositeDisposable?.add(it)
        }

private fun <T> Observable<T>.applyRequestStatus(rxFragment: RxFragment): Observable<T> =
        this.doOnSubscribe { rxFragment.requestInProgress = true }
                .doOnTerminate { rxFragment.requestInProgress = false }

private fun <T> Observable<T>.showLoadingDialog(rxFragment: RxFragment): Observable<T> =
        this.doOnSubscribe { DialogUtils.showProgressDialog(rxFragment.fragmentManager, LOADING_MESSAGE) }
                .doOnTerminate { DialogUtils.hideProgressDialog(rxFragment.fragmentManager) }