Realm + Retrofit + RxJava : Concat 및 SubscribeOn

Question

RxJava 연결 연산자를 사용하는 동안 문제가 발생합니다. 나는이 명 관찰 가능한, 서버 데이터베이스에서 첫 번째를 방출 결과를 가지고 있고 다른 하나는 로컬 데이터베이스의 결과를 방출하고 난 CONCAT : 나는 subscribeOn()를 사용하고 있지 않다 때문에

// Uses a Realm in the UI thread 
Observable<MyResult> remoteObservable = mRemoteDataSource.find(tId); 

// Uses Retrofit 
Observable<MyResult> localObservable = mLocalDataSource.find(tId); 

Observable.concat(localObservable, remoteObservable) 
    .doOnNext(result -> /* Do my stuff */) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .doOnError(throwable -> throwable.printStackTrace()) 
    .subscribe() 

그래서 이것은 나에게 문제가 발생합니다 concatenated observable이 AndroidScheduler.MainThread()에서 실행 중이며 원격을 실행하지 않고 NetworkOnMainThreadException을 실행합니다.

subscribeOn(Schedulers.computation())을 구현하는 경우 Observable이 스레드에서 실행 중이 아니므로 Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created이 나오기 때문에 Realm 인스턴스가 존재합니다.

다른 질문을했지만 유용한 것은 없었습니다. 예를 들어 영역 : https://github.com/realm/realm-java/blob/master/examples/rxJavaExample/src/main/java/io/realm/examples/rxjava/retrofit/RetrofitExample.java을 확인했지만 이상하게도 개조 관찰 항목이 아무 것도 구독하지 않고 작동한다는 것을 알았습니다.

샘플과 코드에서 작동하는 이유는 무엇입니까? 어떠한 제안?

Answer 1

나는 올바른 장소에 subscribeOn()을 사용해야한다고 생각합니다.

// Uses a Realm in the UI thread 
Observable<MyResult> realmObservable = mRealmDataSource.find(tId).subscribeOn(AndroidSchedulers.mainThread()); 

// Uses Retrofit 
Observable<MyResult> retrofitObservable = mRetrofitDataSource.find(tId).subscribeOn(Subscribers.io()); 

Observable.concat(realmObservable, retrofitObservable) 
    .doOnNext(result -> /* Do my stuff */) 
    .subscribeOn(AndroidSchedulers.mainThread()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .doOnError(throwable -> throwable.printStackTrace()) 
    .subscribe() 

문제가 해결되는지 확인해보십시오.

Answer 2

당신은 다음과 같이 당신의 로컬 및 원격 관찰 가능한을 CONCAT 수 있습니다

// Uses a Realm in the UI thread 
Observable<MyResult> remoteObservable = mRemoteDataSource.find(tId); 

// Uses Retrofit 
Observable<MyResult> localObservable = mLocalDataSource.find(tId); 

Observable.concat(localObservable, remoteObservable).first() 
       .map(new Func1<MyResult, MyResult>() { 
        @Override 
        public myResult call(MyResult result) { 
         if (result == null) { 
          throw new IllegalArgumentException(); 
         } 
         return result; 
        } 
       }); 

그리고 아래와 같이 등록 :

CompositeSubscription mCompositeSubscription = new CompositeSubscription(); 
final Subscription subscription = mRepo.find(tId 
       .subscribeOn(Schedulers.io()) 
       .observeOn(AndroidSchedulers.mainThread()) 
       .subscribe(new Observer<MyResult>() { 
        @Override 
        public void onCompleted() { 
         // Completed 
        } 

        @Override 
        public void onError(Throwable e) { 
         // onError 
        } 

        @Override 
        public void onNext(MyResult result) { 
         //onSuccess 
        } 
       }); 
mCompositeSubscription.add(subscription); 

당신은 RxJava + 개조 + 영역이 REPO를 확인할 수 있습니다 https://github.com/savepopulation/wikilight

행운을 빈다!