Nicht zustellbarException beim Aufruf von onError von ObservableEmitter in RXjava2

Question

Ich habe eine Methode, die einen Emitter wie unten erstellt, gibt es ein Problem (vielleicht ist es normales Verhalten) mit dem Aufruf von onError in Retrofit Callback. Ich habe UndeliverableException erhalten, wenn ich versuche, OnError aufzurufen.

Ich kann das lösen, indem ich subscriber.isDiposed() überprüfe, indem ich mich frage, wie onError coz anrufen kann, muss ich meine UI-Ebene benachrichtigen.

• Zusatz 1

--> RxJava2CallAdapterFactoryalready implemented 
private static Retrofit.Builder builderSwift = new Retrofit.Builder() 
      .baseUrl(URL_SWIFT) 
      .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) 
      .addConverterFactory(GsonConverterFactory.create()) 
      .addConverterFactory(new ToStringConverterFactory()); 

--> When i added below code to application class app won't crash 
--> but i get java.lang.exception instead of my custom exception 

RxJavaPlugins.setErrorHandler(Functions<Throwable>emptyConsumer()); 

@Override 
    public void onFileUploadError(Throwable e) { 
     Log.d(TAG, "onFileUploadError: " + e.getMessage()); 
    } 

public Observable<UploadResponseBean> upload(final UploadRequestBean uploadRequestBean, final File file) { 
    return Observable.create(new ObservableOnSubscribe<UploadResponseBean>() { 
     @Override 
     public void subscribe(@NonNull final ObservableEmitter<UploadResponseBean> subscriber) throws Exception { 

       // ---> There are no problem with subscriber while calling onError   

       // ---> Retrofit2 service request 
       ftsService.upload(token, uploadRequestBean, body).enqueue(new Callback<UploadResponseBean>() { 
        @Override 
        public void onResponse(Call<UploadResponseBean> call, Response<UploadResponseBean> response) { 

         if (response.code() == 200){ 
          // ---> calling onNext works properly 
          subscriber.onNext(new UploadResponseBean(response.body().getUrl())); 
         } 
         else{ 
          // ---> calling onError throws UndeliverableException 
          subscriber.onError(new NetworkConnectionException(response.message()));     
         } 
        } 

        @Override 
        public void onFailure(Call call, Throwable t) { 
         subscriber.onError(new NetworkConnectionException(t.getMessage())); 
        } 
       }); 
     } 
    }); 
} 

Answer 1

Das Problem ist, wie Sie sagen, Sie müssen überprüfen, ob Subscriber bereits angeordnet ist, das ist, weil RxJava2 strenger ist in Bezug auf Fehler, die nach Subscriber ausgelöst wurden bereits entsorgt.
RxJava2 liefern diese Art von Fehler zu RxJavaPlugins.onError, die standardmäßig auf Stapel Trace und Aufrufe zum Thread uncaught Exception-Handler zu drucken. Sie können die vollständige Erklärung here lesen.

Nun, was passiert hier, ist, dass Sie wahrscheinlich von dieser Observable abgemeldet (entsorgen), bevor Abfrage wurde durchgeführt und Fehler geliefert und als solche - Sie erhalten die UndeliverableException.

Ich frage mich, wie onError coz aufrufen kann, muss ich meine UI-Ebene benachrichtigen.

Da dies passiert, nachdem Ihre UI abgemeldet wurde die UI sollte nicht kümmern. Im normalen Fluss sollte dieser Fehler korrekt geliefert werden.

Einige allgemeine Punkte in Bezug auf die Implementierung:

• das gleiche Problem bei der onError im Falle passieren Sie vor unsubscribed waren.
• Hier gibt es keine Abbruchlogik (das verursacht dieses Problem), also Anfrage fortsetzen, auch wenn Subscriber abgemeldet.
• , auch wenn Sie diese Logik implementieren werden Sie immer noch dieses Problem auftritt im Fall (ObservableEmitter.setCancellable()/setDisposable() verwenden) Sie abmelden, bevor Anfrage durchgeführt wird - dies wird dazu führen, Löschung und Ihre onFailure Logik onError() und das gleiche Problem nennen wird passieren .
• Wenn Sie einen asynchronen Anruf über Retrofit durchführen, wird das angegebene Abonnement Scheduler die eigentliche Anfrage nicht auf dem Scheduler Thread sondern nur auf dem Abonnement ausführen. Sie können Observable.fromCallable und Retrofit Blocking Call execute verwenden, um mehr Kontrolle über den tatsächlichen Thread-Aufruf zu erhalten ist passiert.

um es zusammenzufassen - Bewachung Anrufe onError() mit ObservableEmitter.isDiposed() ist eine gute Praxis in diesem Fall.
Aber ich denke, die beste Praxis ist, um Retrofit RxJava Call Adapter zu verwenden, so dass Sie Observable, die den Retrofit-Aufruf tun und schon alle diese Überlegungen haben.

Answer 2

Da Version 2.1.1 tryOnError verfügbar ist:

Der Emitter API (wie FlowableEmitter, SingleEmitter, etc.) nun mit einem neuen Verfahren, welche die tryOnError Throwable wenn die Sequenz auszusenden versucht wird nicht storniert/entsorgt. Im Gegensatz zum regulären onError gibt der Downstream keine Ereignisse mehr an, die Methode gibt false zurück und signalisiert keine UndeliverableException.

https://github.com/ReactiveX/RxJava/blob/2.x/CHANGES.md