I'm using Retrofit and RxJava to perform some background tasks. Code looks like this:
public class MyLoader{
public Observable<MyData> getMyData(){
return setupHelper().flatMap(new Func1<MyHelper, Observable<MyData>>() {
@Override
public Observable<MyData> call(MyHelper myHelper) {
return queryData(myHelper);
}
});
}
private Observable<MyData> queryData(MyHelper myHelper){
...
}
private Observable<MyHelper> setupHelper(){
return Observable.create(new Observable.OnSubscribe<MyHelper>() {
@Override
public void call(final Subscriber<? super MyHelper> subscriber) {
try{
MyHelper helper = makeRetrofitCall();//Using Retrofit blocking call to get some data
subscriber.onNext(helper);
subscriber.onCompleted();
}catch(RetrofitError e){
subscriber.onError(e)
}
}
}
}
}
This fails with RetrofitError, due to NetworkOnMainThread
Exception at this line:
MyHelper helper = makeRetrofitCall();//Using Retrofit blocking call to get some data
Subscribing to my Observable:
myLoader.getMyData()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<MyData>() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(MyData inventory) {
}
});
According to Rx documentation flatMap
doesn't operate on any background thread. My question is how do I ensure that the whole getMyData()
method runs in background.
.subscribeOn(Schedulers.io())
intosetupHelper
instead?.subscribeOn()
in the second observable?Observable
directly, you shouldn't need to wrap the call like that. Otherwise, you can also simply useObservable.fromCallable(this::makeRetrofitCall)
(assuming you have functions refs, either via Java 8 or via retrolambda).