0

Вот есть у меня Flowable на который я подписываюсь вот так

private void setListeners()
    {
        Flowable<String> flowable = createTextChangeFlowable();
        mDisposable = flowable.subscribeOn(AndroidSchedulers.mainThread()).subscribe(iS -> updateAnswer());
    }

Метод updateAnswer() работает с UI и получаю вот такую ошибку

FATAL EXCEPTION: RxComputationThreadPool-3
                                                                      Process: com.ewavemobile.supergas, PID: 11531
                                                                      io.reactivex.exceptions.OnErrorNotImplementedException: Only the original thread that created a view hierarchy can touch its views.
                                                                          at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
                                                                          at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
                                                                          at io.reactivex.internal.subscribers.LambdaSubscriber.onError(LambdaSubscriber.java:79)
                                                                          at io.reactivex.internal.subscribers.LambdaSubscriber.onNext(LambdaSubscriber.java:69)
                                                                          at io.reactivex.internal.operators.flowable.FlowableSubscribeOn$SubscribeOnSubscriber.onNext(FlowableSubscribeOn.java:97)
                                                                          at io.reactivex.subscribers.SerializedSubscriber.onNext(SerializedSubscriber.java:100)
                                                                          at io.reactivex.internal.operators.flowable.FlowableDebounceTimed$DebounceTimedSubscriber.emit(FlowableDebounceTimed.java:152)
                                                                          at io.reactivex.internal.operators.flowable.FlowableDebounceTimed$DebounceEmitter.emit(FlowableDebounceTimed.java:188)
                                                                          at io.reactivex.internal.operators.flowable.FlowableDebounceTimed$DebounceEmitter.run(FlowableDebounceTimed.java:183)
                                                                          at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
                                                                          at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
                                                                          at java.util.concurrent.FutureTask.run(FutureTask.java:237)
                                                                          at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
                                                                          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
                                                                          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
                                                                          at java.lang.Thread.run(Thread.java:776)

А если подписываюсь вот так

private void setListeners()
    {
        Flowable<String> flowable = createTextChangeFlowable();
        mDisposable = flowable
                .subscribeOn(Schedulers.single())//
                .observeOn(AndroidSchedulers.mainThread())//
                .subscribe(iS -> updateAnswer());
    }

То все нормально.

Я не пойму почему в первом случае я получаю ошибку если я в этом методе

.subscribeOn(AndroidSchedulers.mainThread())

говорю, что подписаться нужно на UI потоке

Правка

private Flowable<String> createTextChangeFlowable()
    {
        Flowable<String> flowable = Flowable.create(emitter ->
        {
            TextWatcher textWatcher = new TextWatcher()
            {
                @Override
                public void beforeTextChanged(CharSequence s, int start, int count, int after)
                {
                }

                @Override
                public void onTextChanged(CharSequence s, int start, int before, int count)
                {
                }

                @Override
                public void afterTextChanged(Editable s)
                {
                    emitter.onNext(s.toString());
                }
            };

            mEtOpenAnswer.setTag(textWatcher);
            mEtOpenAnswer.addTextChangedListener(textWatcher);
            emitter.setCancellable(() -> mEtOpenAnswer.removeTextChangedListener(textWatcher));

        }, BackpressureStrategy.BUFFER);

        return flowable.debounce(1000, TimeUnit.MILLISECONDS);
    }

1 ответ 1

2

Метод subscribeOn используется для того, что бы сообщить в каком потоке необходимо начать выполнение всей цепочки.

Метод observeOn используется для того, что бы сообщить на каком потоке необходимо выполнить работу цепочки, которая идет после этого оператора. Другими словами он влияет на то, что находится ниже этого оператора.

Ещё следует учитывать что если в цепочке попадутся несколько операторов subscribeOn, то выполнение начнется в потоке, который указан в самом верхнем subscribeOn операторе.

Что касается вашего кода, то где то в createTextChangeFlowable(); есть либо ещё один вызов subscribeOn либо observeOn которые переключают выполнение цепочки на другой поток.

Ниже небольшой пример с комментариями в каком потоке будет выполнен очередной оператор.

loadData(...)                   // Scheduler C
    .observeOn(Scheduler_A)
    .flatMap(...)               // Scheduler A
    .observeOn(Scheduler_B)
    .flatMap(...)               // Scheduler B
    .subscribeOn(Scheduler_C)
    .subscribe()                // Scheduler B

Подробнее и нагляднее это показано в документации

UPD:

Метод debounce(long, TimeUnit) выглядит следующим образом

@CheckReturnValue
@BackpressureSupport(BackpressureKind.ERROR)
@SchedulerSupport(SchedulerSupport.COMPUTATION)
public final Flowable<T> debounce(long timeout, TimeUnit unit) {
    return debounce(timeout, unit, Schedulers.computation());
}

Соответственно все что ниже него будет выполнено в Schedulers.computation()

2
  • Я добавил код создания Flowable , исходя из ответа, могу предположить, что либо этот параметр BackpressureStrategy.BUFFER как то влияет на поток в котором создается либо оператор debounce(1000, TimeUnit.MILLISECONDS) уводит выполнение из UI. Как думаете?
    – Sirop4ik
    19 мар 2018 в 7:27
  • @AlekseyTimoshchenko обновил ответ
    – temq
    19 мар 2018 в 8:00

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.