Osservabili corse sul thread principale anche se subscribeOn() viene chiamato su un altro thread

Question

Ho avuto un problema strano in una delle mie attività. Quando torno da una foto o un video, nel mio onActivityResult sto visualizzando una finestra di dialogo che consente all'utente di nominare la telecamera. Una volta che l'utente preme OK, invio onNext() a un oggetto con il nome del file richiesto che copia il file (e mostra la finestra di avanzamento).

Per qualche motivo la funzione map() che fa la copia viene sempre chiamata sul thread principale, anche se chiamo 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); 
       } 
      }); 
} 

Modifica della subscribeOn(Schedulers.io()) chiamata observeOn(Schedulers.io()) risolto il problema. Ancora mi piacerebbe sapere perché non ha funzionato ...

Answer 1

subscribeOn e observeOn sono gli operatori per lo più confusi che ci sono. Il primo si assicura che gli effetti collaterali dell'abbonamento avvengano sullo schedulatore specificato (thread), ma ciò non significa che i valori appariranno anche su quel thread.

Ad esempio, se il tuo Observer apre una connessione di rete quando ci si iscrive ad esso, non si vuole che venga eseguito sul thread principale, quindi è necessario subscribeOn per specificare dove tale abbonamento e quindi la connessione di rete sarà creato.

Quando i dati arrivano finalmente, il thread di emissione può essere qualsiasi cosa, uno degli scheduler o uno sfondo vecchio thread normale. Dal momento che non conosciamo o non ci piace quel thread, vogliamo spostare l'osservazione dei dati su un altro thread. Questo è ciò che osservaOn fa: assicura che gli operatori dopo di esso eseguano la loro logica onNext sullo scheduler specificato. Gli sviluppatori di Android lo usano già per spostare l'osservazione dei valori sul thread principale.

Ciò che raramente ha spiegato se è ciò che accade quando si vuole un po 'di calcolo in più fuori il thread principale prima che le terre risultato finale sul thread principale di nuovo: uso multiplo observeOn operatori:

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