Obserwowalne działa na głównym wątku, mimo że subscribeOn() jest wywoływana w innym wątku

Question

Mam dziwny problem w jednym z moich działań. Po powrocie z robienia zdjęcia/wideo w moim onActivityResult wyświetlam okno dialogowe, które pozwala użytkownikowi nazwać kamerę. Po naciśnięciu przycisku OK, wysyłam onNext() do tematu z żądaną nazwą pliku, który kopiuje plik (i pokazuje okno dialogowe postępu).

Z jakiegoś powodu funkcja map(), która wykonuje kopię, jest zawsze wywoływana w głównym wątku, mimo że nazywam ją subscribeOn(Schedulers.io()).

@Override 
protected void onActivityResult(final int requestCode, int resultCode, Intent intent) { 
    ... 

    final PublishSubject<String> subject = PublishSubject.create();` 

    mSubscription = subject 
      .subscribeOn(Schedulers.io()) 
      .map(new Func1<String, String>() { 
       @Override 
       public String call(String fileName) { 
        Log.I.d(TAG,"map"); 
        return doSomeIOHeavyFuncition(); 
       } 
      }) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(new Action1<String>() { 
       @Override 
       public void call(final String fullPath) { 
        Log.d(TAG,"onNext"); 
        doSomethingOnUI(fullPath); 

        subject.onCompleted(); 
       } 
      }, new Action1<Throwable>() { 
       @Override 
       public void call(Throwable throwable) { 
        ... 
       } 
      }, new Action0() { 
       @Override 
       public void call() { 
        ... 
       } 
      }); 

    final AlertDialog dialog = new AlertDialog.Builder 
    .... 
    .create() 
      .show(); 

    dialog.getButton(DialogInterface.BUTTON_POSITIVE) 
      .setOnClickListener(new View.OnClickListener() { 
       @Override 
       public void onClick(View view) { 
        String someString = getStringFromDialog(dialog); 

        dialog.dismiss(); 
        InputMethodManager imm = (InputMethodManager) getSystemService(Context.INPUT_METHOD_SERVICE); 
        imm.hideSoftInputFromWindow(input.getWindowToken(), 0); 

        showProgressDialog(); 
        subject.onNext(someString); 
       } 
      }); 
} 

Zmiana subscribeOn(Schedulers.io()) wezwanie do observeOn(Schedulers.io()) rozwiązać ten problem. Nadal chciałbym wiedzieć, dlaczego to nie działa ...

Answer 1

subscribeOn i observeOn jest głównie mylić operatorów istnieją. Ten pierwszy zapewnia, że ​​efekty subskrypcji występują w określonym harmonogramie (wątku), ale to nie znaczy, że wartości pojawią się również w tym wątku.

Na przykład, jeśli Twój obserwator otwiera połączenie sieciowe, gdy się go subskrybuje, nie chcesz, aby działał w głównym wątku, w związku z tym potrzebujesz subskrybowania, aby określić miejsce, w którym subskrypcja, a tym samym połączenie sieciowe, będzie stworzony.

Po otrzymaniu danych wątek emitujący może być dowolnym elementem, jednym z planistów lub zwykłym starym wątkiem tła. Ponieważ nie znamy lub nie lubimy tego wątku, chcemy przenieść obserwację danych do innego wątku. Oto co zaobserwuje: upewnia się, że operatorzy po nim wykonają swoją logikę onNext na określonym harmonogramie. Twórcy Androida używają go już do przeniesienia obserwacji wartości z powrotem do głównego wątku.

Co rzadko wyjaśnił jednak, co się dzieje, gdy chcesz jakieś dodatkowe obliczenia poza głównym wątku przed ostatecznym ziem Wynik na głównym wątku jeszcze: Korzystanie z wielu observeOn operatorzy:

source 
.observeOn(Schedulers.computation()) 
.map(v -> heavyCalculation(v)) 
.observeOn(Schedulers.io()) 
.doOnNext(v -> { saveToDB(v); }) 
.observeOn(AndroidSchedulers.mainThread()) 
...