Hot questions for Using RxJava 2 in interrupted exception

Top Java Programmings / RxJava 2 / interrupted exception

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:

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);
});

Question:

Use implementation 'io.reactivex.rxjava2:rxjava:2.1.9'

I am trying parsing withe rxJava. Parse long json data. So my parse take time. But if user leave screen my parsing need finish work but instead my app crashes. Method for parsing:

    override fun restore(): Observable<List<Pair<String, String>?>> {
return backupRemoteSource.getBackup()
                .flatMap { urlBackup ->
                    Observable.create<Boolean> { emitter ->
                        var isRestore = true
                        try {
                            val url = URL(urlBackup)
                            url.openConnection()
                            InputStreamReader(url.openStream(), "UTF-8").use {
                                val jsonReader = JsonReader(it)

                                jsonReader.beginArray()

                                var tour: Tour? = null

                                /* tours */
                                while (jsonReader.hasNext() ) {

                                    /* tour item*/
                                    jsonReader.beginObject()
                                    while (jsonReader.hasNext() && isRestore) {

                                        val name = jsonReader.nextName()

                                        when (name) {

                                            "tourInfo" -> {

                                                tour = tourMapper.fromRx(gsonParser.fromJson<TourBackup>(jsonReader, TourBackup::class.java))
                                                        .flatMap {


                                                            tourLocalSource.save(it)
                                                        }.blockingFirst()

                                                Log.i(tag, "\n@saved tour")
                                            }

                                      jsonReader.endArray()
                                            }
                                        }

                                    }
                                    jsonReader.endObject()
                                }
                                jsonReader.endArray()

                                emitter.onNext(true)
                            }
                        } catch (error: InterruptedException) {
                            emitter.onError(BackupException(error.message ?: "unknown exception"))
                        } catch (error: NoSuchFileException) {
                            emitter.onError(BackupException(error.message ?: "unknown exception"))
                        } catch (error: IOException) {
                            emitter.onError(BackupException(error.message ?: "unknown exception"))
                        }

                        emitter.setCancellable({
                            Log.d(tag,"cancel restore")
                            isRestore = false
                        })
                    }

                }
....}

then i call:

fun restore() {

    disposables.add(backUpRepo.restore()
            .compose(RxUtils.ioToMainTransformer())
            .subscribe()
}

but when calling during long data parsing:

disposables.dispose()

i get crash:

  04-23 14:57:27.456 28001-28063/com.jellyworkz.udark.debug E/AndroidRuntime: FATAL EXCEPTION: RxCachedThreadScheduler-1
    Process: com.jellyworkz.udark.debug, PID: 28001
    io.reactivex.exceptions.UndeliverableException: java.lang.RuntimeException: java.lang.InterruptedException


....

    Caused by: java.lang.RuntimeException: java.lang.InterruptedException
        at io.reactivex.internal.util.ExceptionHelper.wrapOrThrow(ExceptionHelper.java:45)
        at io.reactivex.internal.observers.BlockingBaseObserver.blockingGet(BlockingBaseObserver.java:74)
        at io.reactivex.Observable.blockingFirst(Observable.java:4987)
        at com.jellyworkz.udark.backup.source.BackupRepositoryImpl$restore$1$1.subscribe(BackupRepositoryImpl.kt:113)

Where do i wrong?


Answer:

Just used emitter.tryOnError

   try{

    ....

    } catch (error: InterruptedException) {
         emitter.tryOnError(BackupException(error.message ?: "unknown exception"))
    }

see more details