NetworkOnMainThreadException wenn Anruf von ConnectableFlowable verbinden

Question

Ich habe Probleme Umgang mit ConnectableFlowable und Themen mit:

Mein Nachrüst-Dienst gibt ein ConnectableFlowable:

service.searchBeers(config.getKey(), name) 
         .map(Mappers.SEARCH_MAPPER).publish(); 

In meiner Interactor Klasse I bis sie abonniert haben Fügen Sie Daten zu einer Firebase-Datenbank hinzu, und dann habe ich ConnectableFlowable an den Präsentator zurückgegeben. Ich habe die Methoden subscribeOn, unsubscribeOn und observeOn, die IO-Thread übergeben, aufgerufen, da die Firebase-Einfügungen nicht im Hauptthread ausgeführt werden müssen.

@Override 
    public Flowable<LocalType<List<Beer>>> searchBeers(String query) { 
     ConnectableFlowable<LocalType<List<Beer>>> connectableFlowable = (ConnectableFlowable<LocalType<List<Beer>>>) 
       remote.search(query); 

     connectableFlowable.unsubscribeOn(schedulerProvider.io()).subscribeOn(schedulerProvider.io()).observeOn 
       (schedulerProvider.io()).subscribe(onNext, 
       onError); 

     return connectableFlowable; 
    } 

In meinem Presenter Ich habe es abonniert die Ansicht zu aktualisieren, so nenne ich observeOn und übergeben schedulerProvider.ui() (das ist ein Wrapper Mainthread).

@Override 
public void search(String query) { 

    ConnectableFlowable<LocalType<List<Beer>>> flowable = (ConnectableFlowable<LocalType<List<Beer>>>) 
      searchInteractor.searchBeers(query); 
    flowable.subscribeOn(schedulerProvider.io()).observeOn(schedulerProvider.ui()).unsubscribeOn 
      (schedulerProvider.io()) 
      .subscribe(
        beerListData -> searchView.showSearchResult(beerListData.getData()), 
        error -> searchView.showSearchError(error) 
      ); 

    flowable.connect(); 
} 

Aber wenn es nennt flowable.connect es eine NetworkOnMainThreadException aufwirft:

E/SearchActivity: showSearchError: null 
        android.os.NetworkOnMainThreadException 
         at android.os.StrictMode$AndroidBlockGuardPolicy.onNetwork(StrictMode.java:1303) 
         at java.net.Inet6AddressImpl.lookupHostByName(Inet6AddressImpl.java:86) 
         at java.net.Inet6AddressImpl.lookupAllHostAddr(Inet6AddressImpl.java:74) 
         at java.net.InetAddress.getAllByName(InetAddress.java:752) 
         at okhttp3.Dns$1.lookup(Dns.java:39) 
         at okhttp3.internal.connection.RouteSelector.resetNextInetSocketAddress(RouteSelector.java:171) 
         at okhttp3.internal.connection.RouteSelector.nextProxy(RouteSelector.java:137) 
         at okhttp3.internal.connection.RouteSelector.next(RouteSelector.java:82) 
         at okhttp3.internal.connection.StreamAllocation.findConnection(StreamAllocation.java:171) 
         at okhttp3.internal.connection.StreamAllocation.findHealthyConnection(StreamAllocation.java:121) 
         at okhttp3.internal.connection.StreamAllocation.newStream(StreamAllocation.java:100) 
         at okhttp3.internal.connection.ConnectInterceptor.intercept(ConnectInterceptor.java:42) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) 
         at okhttp3.internal.cache.CacheInterceptor.intercept(CacheInterceptor.java:93) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) 
         at okhttp3.internal.http.BridgeInterceptor.intercept(BridgeInterceptor.java:93) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) 
         at okhttp3.internal.http.RetryAndFollowUpInterceptor.intercept(RetryAndFollowUpInterceptor.java:120) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:92) 
         at okhttp3.internal.http.RealInterceptorChain.proceed(RealInterceptorChain.java:67) 
         at okhttp3.RealCall.getResponseWithInterceptorChain(RealCall.java:185) 
         at okhttp3.RealCall.execute(RealCall.java:69) 
         at retrofit2.OkHttpCall.execute(OkHttpCall.java:180) 
         at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:41) 
         at io.reactivex.Observable.subscribe(Observable.java:10910) 
         at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34) 
         at io.reactivex.Observable.subscribe(Observable.java:10910) 
         at io.reactivex.internal.operators.flowable.FlowableFromObservable.subscribeActual(FlowableFromObservable.java:29) 
         at io.reactivex.Flowable.subscribe(Flowable.java:12994) 
         at io.reactivex.internal.operators.flowable.FlowableOnBackpressureLatest.subscribeActual(FlowableOnBackpressureLatest.java:32) 
         at io.reactivex.Flowable.subscribe(Flowable.java:12994) 
         at io.reactivex.internal.operators.flowable.FlowableMap.subscribeActual(FlowableMap.java:38) 
         at io.reactivex.Flowable.subscribe(Flowable.java:12994) 
         at io.reactivex.internal.operators.flowable.FlowablePublish.connect(FlowablePublish.java:130) 
         at io.reactivex.flowables.ConnectableFlowable.connect(ConnectableFlowable.java:64) 
         at com.github.alexpfx.udacity.beercollection.beer.search.DefaultSearchPresenter.search(DefaultSearchPresenter.java:50) 
         at com.github.alexpfx.udacity.beercollection.SearchActivity.actionSearch(SearchActivity.java:63) 
         at com.github.alexpfx.udacity.beercollection.SearchActivity_ViewBinding$1.doClick(SearchActivity_ViewBinding.java:35) 
         at butterknife.internal.DebouncingOnClickListener.onClick(DebouncingOnClickListener.java:22) 
         at android.view.View.performClick(View.java:6261) 
         at android.view.View$PerformClick.run(View.java:23748) 
         at android.os.Handler.handleCallback(Handler.java:751) 
         at android.os.Handler.dispatchMessage(Handler.java:95) 
         at android.os.Looper.loop(Looper.java:154) 
         at android.app.ActivityThread.main(ActivityThread.java:6776) 
         at java.lang.reflect.Method.invoke(Native Method) 
         at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:1496) 
         at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:1386) 

ich nicht erkennen, was falsch läuft, die abonnieren da nicht in Haupt-Thread ausgeführt wird.

- My SchedulerProvider Klasse:

@Singleton 
public class AndroidSchedulerProvider implements SchedulerProvider { 

    @Inject 
    public AndroidSchedulerProvider() { 
    } 

    @Override 
    public Scheduler computation() { 
     return Schedulers.computation(); 
    } 

    @Override 
    public Scheduler io() { 
     return Schedulers.io(); 
    } 

    @Override 
    public Scheduler ui() { 
     return AndroidSchedulers.mainThread(); 
    } 
} 

Answer 1

sollten Sie die die publish().subscribeOn(schedulerProvider.io())vor setzen, so dass die zugrunde liegenden service.searchBeersFlowable auf io Thread abonnieren werden, wenn Sie die subscribeOn nach veröffentlichen setzen sie die Auswirkungen auf Abonnement für die ConnectableFlowable selbst nur.

service.searchBeers(config.getKey(), name) 
      .map(Mappers.SEARCH_MAPPER) 
      .subscribeOn(schedulerProvider.io()) 
      .publish();