FATALE AUSNAHME: RxCachedThreadScheduler-1, wenn Trigger absetzen. Warum?

Question

ich folgende RxJava 2-Code (in Kotlin) haben, die eine beobachtbare haben, dass

disposable = Observable.create<String>({ 
    subscriber -> 
      try { 
       Thread.sleep(2000) 
       subscriber.onNext("Test") 
       subscriber.onComplete() 
      } catch (exception: Exception) { 
       subscriber.onError(exception) 
      } 
}).subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe({ result -> Log.d("Test", "Completed $result") }, 
      { error -> Log.e("Test", "Completed ${error.message}") }) 

Während es noch Thread.sleep(2000), ich disposable?.dispose() Anruf durchführen, wird es Fehler aus

FATAL EXCEPTION: RxCachedThreadScheduler-1 
Process: com.elyeproj.rxstate, PID: 10202 
java.lang.InterruptedException 
    at java.lang.Thread.sleep(Native Method) 
    at java.lang.Thread.sleep(Thread.java:371) 
    at java.lang.Thread.sleep(Thread.java:313) 
    at presenter.MainPresenter$loadData$1.subscribe(MainPresenter.kt:41) 
    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40) 

Ich erwarte, Die dispose würde helfen, die Operation still oder höchstens zu beenden, den Fehler mit dem Log.e auf dem Abonnement zu fangen. Es stürzt jedoch nur ab, wie in der obigen Fehlermeldung.

Warum ist die Ausnahme entkommen? Ist nicht dispose angenommen, die gesamte Operation still und ohne Absturz abzubrechen?

Answer 1

Es ist eine Kombination von Faktoren hier:

1. dispose eines Stroms, der verwendete subscribeOn auch verfügt über den Thread verwendet. Dies beinhaltet auch den Aufruf Thread.interrupt() bei Verwendung von Schedulers.io(). Dies verursacht die Ausnahme in Ihrem Fall.
2. InterruptedException ist ein Exception geworfen von Thread.sleep, so wird es von Ihrem Code gefangen und an onError wie jede andere Ausnahme weitergegeben.

3. Aufruf onError nach dispose leitet den Fehler an die globale Fehlerbehandlungsroutine aufgrund RxJava2 der Politik der NIE wegwerfen Fehler. Um dies zu umgehen, überprüfen Sie subscriber.isDisposed() vor dem Aufruf onError oder verwenden Sie RxJava 2.1.1 neue subscriber.tryOnError.

   if (!subscriber.isDisposed()) { 
       subscriber.onError(exception) 
   }