java - How to handle dispose in RxJava without InterruptedException

Question

In the below code snipped when dispose() is called, then the emitter thread is interrupted (InterruptedException is thrown out of sleep method).

    Observable<Integer> obs = Observable.create(emitter -> {
        for (int i = 0; i < 10; i++) {
            if (emitter.isDisposed()) {
                System.out.println("> exiting.");
                emitter.onComplete();
                return;
            }

            emitter.onNext(i);
            System.out.println("> calculation = " + i);


            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        emitter.onComplete();
    });

    Disposable disposable = obs
            .subscribeOn(Schedulers.computation())
            .subscribe(System.out::println);

    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    disposable.dispose();

From debugging session I see that the interrupt origins from FutureTask which is cancelled during disposal. In there, the thread that is calling dispose() is checked against runner thread, and if it does not match, the emitter is interrupted. The thread is different since I used computation Scheduler.

Is there any way to make dispose not interrupt such emitter or is it how this actually should always be handled? An issue I see with this approach is when I would have an interruptible operation (simulated here by sleep) that I would want to complete normally before calling onComplete().

Answer 1

Please refer to What's different in 2.0 - Error handling.

One important design requirement for 2.x is that no Throwable errors should be swallowed. This means errors that can't be emitted because the downstream's lifecycle already reached its terminal state or the downstream cancelled a sequence which was about to emit an error.

So you can either wrap everything inside a try/catch and properly emit the error:

Observable<Integer> obs = Observable.create(emitter -> {
   try {
      // ...
   } catch (InterruptedException ex) {
      // check if the interrupt is due to cancellation
      // if so, no need to signal the InterruptedException
      if (!disposable.isDisposed()) {
         observer.onError(ex);
      }
   }
});

or setup a global error consumer to ignore it:

RxJavaPlugins.setErrorHandler(e -> {
    // ..
    if (e instanceof InterruptedException) {
        // fine, some blocking code was interrupted by a dispose call
        return;
    }
    // ...
    Log.warning("Undeliverable exception received, not sure what to do", e);
});