Rxでsubscribe後に一定時間以上経過しても結果が返っていなければローティングを表示する

MainActivity.java

package app.roana0229.org.rxandroidsample.view;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.view.View;
import android.widget.TextView;

import java.util.concurrent.TimeUnit;

import app.roana0229.org.rxandroidsample.R;
import io.reactivex.Completable;
import io.reactivex.Single;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.observers.DisposableSingleObserver;
import io.reactivex.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {

    final private Completable loadingCompletable = Completable.timer(500, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnComplete(new Action() {
                @Override
                public void run() throws Exception {
                    addText("show loading");
                    findViewById(R.id.loading).setVisibility(View.VISIBLE);
                }
            })
            .observeOn(Schedulers.newThread())
            .andThen(Completable.never())
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread())
            .doOnDispose(new Action() {
                @Override
                public void run() throws Exception {
                    addText("hide loading");
                    findViewById(R.id.loading).setVisibility(View.GONE);
                }
            });

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);

        findViewById(R.id.button_left).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                clear();

                final Disposable loadingDisposable = loadingCompletable.subscribe();
                Single.just("request") // ローカルから取得する場合
                        .subscribeOn(Schedulers.newThread())
                        .observeOn(AndroidSchedulers.mainThread())
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                addText("subscribe");
                            }
                        })
                        .doFinally(new Action() {
                            @Override
                            public void run() throws Exception {
                                addText("finally");
                                loadingDisposable.dispose();
                            }
                        })
                        .subscribeWith(new DisposableSingleObserver<String>() {
                            @Override
                            public void onSuccess(String s) {
                                addText("onSuccess");
                            }

                            @Override
                            public void onError(Throwable e) {
                                addText("onError");
                            }
                        });
            }
        });
        findViewById(R.id.button_right).setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                clear();

                final Disposable loadingDisposable = loadingCompletable.subscribe();
                Single.just("request") // リモートから取得する場合
                        .delay(2, TimeUnit.SECONDS) // 実際は同期か非同期かは隠蔽されている
                        .subscribeOn(Schedulers.newThread())
                        .observeOn(AndroidSchedulers.mainThread())
                        .doOnSubscribe(new Consumer<Disposable>() {
                            @Override
                            public void accept(Disposable disposable) throws Exception {
                                addText("subscribe");
                            }
                        })
                        .doFinally(new Action() {
                            @Override
                            public void run() throws Exception {
                                addText("finally");
                                loadingDisposable.dispose();
                            }
                        })
                        .subscribeWith(new DisposableSingleObserver<String>() {
                            @Override
                            public void onSuccess(String s) {
                                addText("onSuccess");
                            }

                            @Override
                            public void onError(Throwable e) {
                                addText("onError");
                            }
                        });
            }
        });
    }

    private void addText(String s) {
        TextView textView = (TextView) findViewById(R.id.text);
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(textView.getText().toString());
        if (stringBuilder.length() > 0) {
            stringBuilder.append("\n");
        }
        stringBuilder.append("> ");
        stringBuilder.append(s);
        textView.setText(stringBuilder.toString());
    }

    private void clear() {
        TextView textView = (TextView) findViewById(R.id.text);
        textView.setText("");
    }
}

実行した時の動作

結局Rxは全く関係なく、ローディング表示用のUtilityクラスを作って、showWithDelay(delayTime), hide() みたいにすれば同じな気がする。