RxAndroid ViewObservable NetworkOnMainThreadException

Question

У меня есть Button, из которого я создаю Observable<OnClickEvent>.

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

Этот пример бросает android.os.NetworkOnMainThreadException:

Observable<OnClickEvent> networkButtonObservable = ViewObservable.clicks(testNetworkButton); 
networkButtonObservable 
    .map(new Func1<OnClickEvent, List<String>>() { 
      @Override 
      public List<String> call(OnClickEvent onClickEvent) { 
       return TestAPI.getTestService().fetchTestResponse(); 
      } 
     } 
    ) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Action1<Object>() { 
        @Override 
        public void call(Object o) {Log.w("Final result: " + o); 
        } 
       } 
    ); 

Так что я стараюсь из другого потока.

Следующие броски rx.exceptions.OnErrorNotImplementedException: Observers must subscribe from the main UI thread, but was Thread[RxNewThreadScheduler-1,5,main]:

networkButtonObservable 
    .subscribeOn(Schedulers.newThread()) 
    .map(new Func1<OnClickEvent, List<String>>() { 
      @Override 
      public List<String> call(OnClickEvent onClickEvent) { 
       return TestAPI.getTestService().fetchTestResponse(); 
      } 
     } 
    ) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Action1<Object>() { 
        @Override 
        public void call(Object o) {Log.w("Final result: " + o); 
        } 
       } 
    ); 

Ok .. Теперь я стараюсь с .debounce() в начале:

networkButtonObservable 
    .debounce(10, TimeUnit.MILLISECONDS) 
    .map(new Func1<OnClickEvent, List<String>>() { 
      @Override 
      public List<String> call(OnClickEvent onClickEvent) { 
       return TestAPI.getTestService().fetchTestResponse(); 
      } 
     } 
    ) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Action1<Object>() { 
        @Override 
        public void call(Object o) {Log.w("Final result: " + o); 
        } 
       } 
    ); 

И это удается.

Очевидно, что я не люблю добавлять задержки в свой код, поэтому я пытаюсь выяснить, что происходит, по-нити. Почему первый пример не выполняет код внутри .map() в фоновом потоке?

Или что мне здесь не хватает?

--- Обновление

я изменить TestApi вернуть наблюдаемый и изменить первый вызов в networkButtonObservable к .flatMap(). Это также работает правильно. Но я до сих пор не знаю, почему исходный способ использования .map() должен завершиться неудачей.

networkButtonObservable 
    .flatMap(new Func1<OnClickEvent, Observable<?>>() { 
     @Override 
     public Observable<?> call(OnClickEvent onClickEvent) { 
      return TestAPI.getTestService().fetchTestResponseObservable(); 
     } 
    }) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Action1<Object>() { 
        @Override 
        public void call(Object o) {Log.w("Final result: " + o); 
        } 
       } 
    ); 

Answer 1

Я не эксперт в Android, но на основе сообщений об ошибках, я думаю, что вам нужно, чтобы подпрыгнуть значение между основным потоком и фоновым потоком. Обычно, Android примеры показывают, добавить пару subscribeOn/observeOn для вашей потоковой обработки:

Observable.just(1) 
.map(v -> doBackgroundWork()) 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(v -> {}); 

, но в этих случаях, «источник», как правило, простуда наблюдаемым вы контролируете.

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

В этом случае, вы можете использовать observeOn несколько раз:

networkButtonObservable 
.subscribeOn(AndroidSchedulers.mainThread()) // just in case 
.observeOn(Schedulers.io()) 
.map(v -> TestAPI.getTestService().fetchTestResponse()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(v -> updateGUI(v)); 

я думаю fetchTestResponseObservable имеет свой собственный subscribeOn или observeOn применяется к нему, так что не бросает исключение сети.

Также я хотел бы упомянуть, что использование нескольких subscribeOn функционально эквивалентно использованию только одного, наиболее близкого к источнику излучения, но технически оно будет запускать неиспользуемые ресурсы потоковой передачи. Однако использование нескольких потоков observeOn в потоке имеет актуальность, поскольку вы можете осмысленно «конвейерно» обрабатывать потоки между потоками с ними.