Verwalten von Thread-Scheduler, wenn concat Observablen

Question

ich einen Code wie dieses:

service.getUserById(10) 
    .subscribeOn(Schedulers.newThread()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .concatMap(getFullUserFromDto()) 
    .subscribe(doSomehingWithUser()); 

private Func1<UserDto, Observable<User>> getFullUserFromDto() { 
    return new Func1<UserDto, Observable<User>>() { 
     @Override 
     public Observable<User> call(final UserDto dto) { 
      return dao.getUserById(dto.getUserId()); 
     } 
    }; 
} 

und in meinem DAO, ich habe:

public Observable<User> getUserById(final Long id) { 
     return api.getUserById(id).map(//more things... 
} 

Hinweis gibt es zwei Ebenen der "Verkettung": service - >dao ->api. Methode api.getUserById(id) einen Netzwerkanruf tätigen.

Ich bekomme NetworkOnMainThreadException Fehler. Warum? Ich benutze und subscribeOn und observeOn Betreiber, aber es scheint, dass es nicht auf die "endgültige" gebaut Observable angewendet wird.

Wenn ich diese Operatoren in dem API-Aufruf verwenden, in der DAO, es funktioniert:

return api.getUserById(id) 
    .subscribeOn(Schedulers.newThread()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .map(//more things... 

Gibt es eine Möglichkeit nur einmal in der "root" Observable zu benutzen?

Answer 1

So abonniert concatMap auf Observables. Welcher Thread wird für diesen Vorgang verwendet? Nun, der Thread, der onNext für die concatMat aufgerufen hat, vorausgesetzt, dass es Threads/Scheduler nicht ändert. So sollte eine einfache Umsetzung dabei helfen:

service.getUserById(10) 
    .subscribeOn(Schedulers.newThread()) 
    .concatMap(getFullUserFromDto()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(doSomehingWithUser()); 

ich auch Schedulers.io verwenden würde vorschlagen(), wie es Fäden wiederverwenden.

Answer 2

Kurze Antwort: Verwenden Sie observeOn vor verkettete Operationen auf dem Disponenten kontrollieren sie ausgeführt werden:

service.getUserById(10) 
    .subscribeOn(Schedulers.newThread()) 
    .observeOn(Schedulers.io()) 
    .concatMap(getFullUserFromDto()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(doSomehingWithUser()); 

In dem obigen Beispiel wird .concatMap in Schedulers.io()

Weitere Details ausgeführt werden hier gefunden werden kann: http://tomstechnicalblog.blogspot.com/2016/02/rxjava-understanding-observeon-and.html