Как вы показываете spinner, если наблюдаемый RxJava длится долго?

Я чувствую, что кто-то должен был это пробовать, но я не могу найти хороший способ сделать что-то, если наблюдаемое займет много времени.

Это поток, который я хочу.

Start a search.
If the search takes longer than some time,
    show a spinner or show progress bar.
When finished do subscription action and hide spinner/progress bar.

Ближайший, о котором я могу думать, похож на Zip

manager.search(searchTerm)
       .zip(Observable.Timer(1, TimeUnit.SECONDS))
       .subscribe(
           // if the search is non null then we are good
           // if the long time is non 0 we need to show spinner
       );

Есть ли что-то лучше? Я пробовал весь день без успеха. В идеальном мире я чувствую, что хочу что-то вроде

manager.search(searchTerm)
       .timeout(i -> /* do timeout stuff */, 1, TimeUnit.SECONDS)
       .subscribe(item -> /* do search result stuff */);

Ответ 1

Вы можете сделать это, опубликовав поиск. Наблюдаемый через таймаут:

Observable<Integer> source = Observable.just(1).delay(5, TimeUnit.SECONDS);

source
.doOnSubscribe(() -> System.out.println("Starting"))
.publish(o -> 
    o.timeout(1, TimeUnit.SECONDS, Observable.<Integer>fromCallable(() -> {
        System.out.println("Spinning...");
        return null;
    })).ignoreElements().mergeWith(o)
)
.toBlocking()
.subscribe(v -> {
    System.out.println("Hide spinner if shown.");
    System.out.println(v);
});

Это работает, разбивая источник на две горячие полосы: первая полоса будет запускать оператор timeout, который при истечении времени запускает другое Наблюдаемое с побочным эффектом, который показывает вращающееся управление. Один из способов - использовать fromCallable для этого и игнорировать его результат (это также позволяет избежать дублирования). Вторая полоса будет неизменной и объединена с линией ожидания, чтобы доставить фактическое значение.

Ответ 2

Сегодня я нашел немного странное, но рабочее решение. Идея в том, чтобы использовать интервал вместо таймера.

    fun <T> base_delayed_progress_observable(source: Observable<T>): Observable<T>
    {
        val timer = Observable.interval(100, TimeUnit.MILLISECONDS) //Creates observable that will emit Long++ each 100 miliseconds
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(
                    {
                        if (it == 10L)//Here we check current timer value. For example here i check if it is 1 second gone (100 miliseconds * 10 = 1 second)
                        {
                            //here we put all we need to show progress/spinner an so on
                        }
                    })

        return Observable.zip(source, timer,
            BiFunction<T, Long, T> { t1, t2 ->
                //Here we return our original Obervable zipped with timer
                //Timer will be cancelled when our source Observable gets to OnComplete
                [email protected] t1
            }).doFinally(
            {
                //Here we can dismiss all progress dilogs/spinner
            })
    }