RxJava2 amb Operator throws java.lang.InterruptedException

Question

I am working with Rxjava2 ambArray operator. I have created 2 observable for experiment.

Observable 1:

Observable<String> observable1 =
Observable.defer(new Callable<ObservableSource<? extends String>>() {
    @Override
    public ObservableSource<? extends String> call() throws Exception {
        return  Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(final ObservableEmitter<String> emitter) throws Exception {

                long sleepTime = getRandomNumber();
                Log.i(TAG, "Oservable1 : Sleep: " + sleepTime + " Thread: " + Thread.currentThread().getName());

                Thread.sleep(sleepTime);

                emitter.onNext( "From Observable1 Result" );
                emitter.onComplete();

            }
        })
                .subscribeOn(Schedulers.newThread() );
    }
});

Observable 2:

Observable<String> observable2 =
        Observable.defer(new Callable<ObservableSource<? extends String>>() {
            @Override
            public ObservableSource<? extends String> call() throws Exception {
                return Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(final ObservableEmitter<String> emitter) throws Exception {

                        long sleepTime = getRandomNumber();
                        Log.i(TAG, "Oservable2 :  Sleep: " + sleepTime + " Thread: " + Thread.currentThread().getName());

                        Thread.sleep(sleepTime);

                        emitter.onNext("From Observable2 Result" );
                        emitter.onComplete();

                    }
                })
                        .subscribeOn(Schedulers.newThread());
            }
        });

getRandomNumber method:

private long getRandomNumber() {
    return (long) (Math.random() * 100);
}

I am creating a random delay to return a String.

ambArray:

    Observable.ambArray(observable1, observable2)
            .subscribeOn(Schedulers.io())
            .subscribe(new Consumer<String>() {
                @Override
                public void accept(String s) throws Exception {
                    Log.i(TAG, "Result : " + s);
                }
            });

This cause me java.lang.InterruptedException.

When i remove sleep from observables ambArray works fine. And if i create a new Thread manually then use sleep this also works fine.

Re Written Observable with manual Thread:

Observable<String> observable1 =
Observable.defer(new Callable<ObservableSource<? extends String>>() {
    @Override
    public ObservableSource<? extends String> call() throws Exception {
        return  Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(final ObservableEmitter<String> emitter) throws Exception {


                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        long sleepTime = getRandomNumber();
                        Log.i(TAG, "Oservable1 : Sleep: " + sleepTime + " Thread: " + Thread.currentThread().getName());
                        try {
                            Thread.sleep(sleepTime);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            Log.e(TAG, e.getMessage());
                        }
                        emitter.onNext( "From Observable1 Result" );
                        emitter.onComplete();
                    }
                }).start();


            }
        })
               /* .subscribeOn(Schedulers.newThread() )*/;
    }
});

If i use Thread this way in both observable it works fine. Why using Schedulers Thread cause Exception? and Why using Thread manually don't cause any Exception? What's the difference ?

Edit:

Logcat:

     W/System.err: java.lang.InterruptedException
     W/System.err:     at java.lang.Thread.sleep(Native Method)
     W/System.err:     at java.lang.Thread.sleep(Thread.java:1031)
     W/System.err:     at java.lang.Thread.sleep(Thread.java:985)
     W/System.err:     at com.yousuf.shawon.rxandroidsamples.fragments.AmdExampleFragment$5$1.subscribe(AmdExampleFragment.java:306)
     W/System.err:     at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
     W/System.err:     at io.reactivex.Observable.subscribe(Observable.java:11442)
     W/System.err:     at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
     W/System.err:     at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:571)
     W/System.err:     at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
     W/System.err:     at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
     W/System.err:     at java.util.concurrent.FutureTask.run(FutureTask.java:237)
     W/System.err:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:152)
     W/System.err:     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:265)
     W/System.err:     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1112)
     W/System.err:     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:587)
     W/System.err:     at java.lang.Thread.run(Thread.java:818)

akarnokd akarnokd · Accepted Answer · 2018-03-18T12:26:17

The using a Scheduler, the backing thread pool remembers which thread is serving the method that sleeps and can interrupt it upon cancellation. With plain Threads you start, that reference is lost and you haven't wired up those to the cancellation mechanism:

@Override
public void subscribe(final ObservableEmitter<String> emitter) throws Exception {
    Thread t = new Thread(new Runnable() {
        @Override
        public void run() {
            long sleepTime = getRandomNumber();
            Log.i(TAG, "Oservable1 : Sleep: " + sleepTime + 
                  " Thread: " + Thread.currentThread().getName());
            try {
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
                Log.e(TAG, e.getMessage());
            }
            emitter.onNext( "From Observable1 Result" );
            emitter.onComplete();
        }
    });
    t.start();
    emitter.setCancellable(() -> t.interrupt());
}

RxJava2 amb Operator throws java.lang.InterruptedException

1 Answers