Anwenden von Scheduler zweimal in einer beobachtbaren Kette (mit komponieren)

Question

Ich habe mehrere Netzwerkanrufe in meiner App. Ich mag unter Verwendung des compose Operators mit diesem Transformator eine Netzwerkanforderung im IO-Thread ausgeführt:

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

Dies scheint auch so lange zu arbeiten, wie ich nur einzelne Netzwerk-Anrufe hat. Wenn ich sie jedoch wie im folgenden Beispiel verkette, erhalte ich die NetworkInMainThreadException von 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)); 
} 

Meine Idee war, bevor die compose vor dem Aufruf zur vollständigen beobachtbaren Kette angelegt wird, und dass später compose Anrufe „überschreiben“ das Verhalten eines vorherigen würden. Aber tatsächlich sieht es so aus, als ob der observeOn Anruf des ersten compose (observeOn Hauptthreads) den zweiten compose Anruf (subscribeOn der IO-Thread) dominiert. Eine naheliegende Lösung wäre, zwei Versionen von networkCall1 zu haben - eine, die Scheduler anwendet, und eine andere, die dies nicht tut. Dies würde jedoch meinen Code ziemlich ausführlich machen.

Kennen Sie bessere Lösungen? Können Sie das Verhalten des Anwendens von Schedulern in einer beobachtbaren Kette zweimal (mit Compose) erklären?

Bearbeiten: Ich benutze Retrofit mit RxJava für meine Netzwerkanrufe.

Answer 1

Sie können subscribeOn() nur einmal pro Stream verwenden. Wenn Sie es ein zweites Mal benutzen, wird es nichts tun. Als solcher, wenn Sie Ihre beiden Methoden zusammen sind Verkettungs Sie laufen:

observeOn(AndroidSchedulers.mainThread())

, die den Betrieb schaltet auf den Haupt-Thread. Danach bleibt es dort stehen, weil der nächste subscribeOn() effektiv ignoriert wird.

Ich würde vorschlagen, dass Sie tatsächlich Dinge mit Ihrer Kompositionsmethode übermäßig komplizieren. Fügen Sie einfach

subscribeOn(Schedulers.io())

Um sowohl Ihrem Netzwerk Anrufe und dann

observeOn(AndroidSchedulers.mainThread())

nur verwenden, bevor Sie auf dem Haupt-Thread verarbeiten wollen Ergebnisse. Sie würden am Ende mit etwas wie:

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

Wenn Sie wirklich einen observeOn() Anruf auf den einzelnen Netzanruf Methoden haben möchten Sie können. Sie müssten Ihrem chainedCalls()-Verfahren eine zusätzliche observeOn() hinzufügen. Sie können so viele observeOn() Anrufe pro Stream haben, wie Sie möchten. Es wäre etwas wie:

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()); 
}