Applicazione degli scheduler due volte in una catena osservabile (utilizzando la composizione)

Question

Ho più chiamate di rete nella mia app. Mi piace correre una richiesta di rete nel thread IO utilizzando l'operatore di composizione con questo trasformatore:

public static <T> Transformer<T, T> runOnIoThread() 
{ 
    return tObservable -> tObservable.subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()); 
} 

Questo sembra funzionare bene finché ho solo chiamate di rete singoli. Tuttavia, se li concateno come nell'esempio seguente, ricevo il NetworkInMainThreadException di Android.

public Observable<String> networkCall1() 
{ 
    return <NETWORK_CALL_1()> 
      .compose(runOnIoThread()); 
} 

public Observable<String> networkCall2(String input) 
{ 
    return <NETWORK_CALL_2(input)> 
      .compose(runOnIoThread()); 
} 

public Observable<String> chainedCalls() 
{ 
    return networkCall1() 
      .flatMap(result1 -> networkCall2(result1)); 
} 

La mia idea era che prima compose viene applicato alla catena osservabile completa prima della chiamata, e che più tardi sarebbe compose chiamate "sovrascrivere" il comportamento di uno precedente. In realtà, sembra che la chiamata observeOn del primo compose (observeOn thread principale) domini la seconda chiamata compose (subscribeOn il thread IO). Una soluzione ovvia sarebbe avere due versioni di networkCall1 - una che applica gli scheduler e un'altra che non lo fa. Tuttavia, questo renderebbe il mio codice abbastanza dettagliato.

Conosci soluzioni migliori? Puoi spiegare il comportamento di applicare gli scheduler due volte (con compose) in una catena osservabile?

Modifica: Sto usando il retrofit con RxJava per le mie chiamate di rete.

Answer 1

È possibile utilizzare solo subscribeOn() una volta per flusso. Se lo usi una seconda volta, non farà nulla. Come tale quando si sta concatenamento tuoi due metodi insieme si esegue:

observeOn(AndroidSchedulers.mainThread())

che commuta l'operazione sopra al thread principale. Dopodiché rimane lì perché il prossimo subscribeOn() viene effettivamente ignorato.

Vorrei suggerire che si stanno effettivamente complicando le cose con il metodo di composizione. Basta aggiungere

subscribeOn(Schedulers.io())

Per entrambe le chiamate di rete e quindi utilizzare

observeOn(AndroidSchedulers.mainThread())

Poco prima si desidera elaborare risultati sul thread principale. Faresti finisce con qualcosa di simile:

public Observable<String> networkCall1() 
{ 
    return <NETWORK_CALL_1()> 
      .subscribeOn(Schedulers.io); 
} 

public Observable<String> networkCall2(String input) 
{ 
    return <NETWORK_CALL_2(input)> 
      .subscribeOn(Schedulers.io); 
} 

public Observable<String> chainedCalls() 
{ 
    return networkCall1() 
      .flatMap(result1 -> networkCall2(result1)) 
      .observeOn(AndroidSchedulers.mainThread()); 
} 

EDIT

Se davvero vuole avere un observeOn() chiamata sui singoli metodi di chiamata di rete che è possibile. Dovresti aggiungere un ulteriore observeOn() al tuo metodo chainedCalls(). Puoi avere tutte le chiamate observeOn() che desideri per flusso. Sarebbe qualcosa di simile:

public Observable<String> networkCall1() 
{ 
    return <NETWORK_CALL_1()> 
      .subscribeOn(Schedulers.io) 
      .observeOn(AndroidSchedulers.mainThread()); 
} 

public Observable<String> networkCall2(String input) 
{ 
    return <NETWORK_CALL_2(input)> 
      .subscribeOn(Schedulers.io) 
      .observeOn(AndroidSchedulers.mainThread()); 
} 

public Observable<String> chainedCalls() 
{ 
    return networkCall1() 
      .observeOn(Schedulers.io) 
      .flatMap(result1 -> networkCall2(result1)) 
      .observeOn(AndroidSchedulers.mainThread()); 
}