Unsubscribing does not work when using subscribeOn(Schedulers.newThread())

[oasalonen]

I'm having trouble unsubscribing whenever I use Observable.subscribeOn(Schedulers.newThread()) on Android. I'm using rxjava 0.14.3. In my case, my observer class does the following:

Subscription sub = myObject.getData().subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(this);

The getData() method creates an observable, and the onSubscribeFunc generates some mock data using a new thread:

final BooleanSubscription subscription = new BooleanSubscription();
Thread t = new Thread(new Runnable() {
    public void run() {
        // generate data, call onNext()
        // check subscription.isUnsubscribed()
        // sleep a bit, repeat last two steps
    }
});
t.start();
return subscription;

When I unsubscribe the (Composite) subscription I got in the observer, isUnsubscribed() never returns true in the worker thread and BooleanSubscription.unsubscribe() never gets called (breakpoint never triggered). I stepped through the CompositeSubscription.unsubscribe() method, and it seems that it doesn't even contain the BooleanSubscription anywhere in the keySet.

The interesting thing is that if I call subscribeOn(Schedulers.threadPoolForComputation()) or call it using AndroidSchedulers.mainThread(), the unsubscription works correctly. Am I misusing the Schedulers.newThread() method or does the NewThreadScheduler have a bug?

[zsxwing]

Could you provide a unit test to reveal this issue? I tried similar codes but did not find the bug you described.

[peter-tackage]

I've observed similar behaviour using rxjava 0.14.6, also on Android when attempting to unsubscribe from a Subscription created using Schedulers.newThread().schedulePeriodically(...)

The failure is fairly easy to reproduce in the context of the application, although it's inconsistent as often the subscriptions are successfully unsubscribed.

[rupertbates]

This test demonstrates the problem, basically the Observable never receives unsubscribes even though the Observer no longer receives notifications:

package tests;

import org.junit.Test;
import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.concurrency.Schedulers;
import rx.util.functions.Func1;

import java.util.concurrent.TimeUnit;

public class UnsubscribeTest {
    @Test
    public void testUnsubscribe() throws InterruptedException {
        Subscription sub = Observable.interval(1, TimeUnit.SECONDS)
                .map(new Func1<Long, Long>() {
                    @Override
                    public Long call(Long aLong) {
                        System.out.println("generated " + aLong);
                        return aLong;
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.currentThread())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onCompleted() {
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onNext(Long args) {
                        System.out.println("Received " + args);
                    }
                });

        Thread.sleep(3000);
        System.out.println("Calling unsubscribe");
        sub.unsubscribe();
        Thread.sleep(3000);
    }
}

Output is:

generated 0
Received 0
generated 1
Received 1
Calling unsubscribe
generated 2
generated 3
generated 4
generated 5

If I change the subscribeOn thread to Schedulers.threadPoolForIO() then the output is:

generated 0
Received 0
generated 1
Received 1
Calling unsubscribe

[benjchristensen]

I'll take a look.

[benjchristensen]

I think this is fixed in #472

Can someone confirm?

[rupertbates]

Yes looks fixed to me.

[benjchristensen]

Great, I'll release today or tomorrow.

[hazam]

Hello guys - I'm on 0.15.1 and still see the issue, on any scheduler you subscribeOn.
Issue is a little different - I go on getting called onNext on the observer even after unsubscribe(). Only difference is that onComplete is not called.

[akarnokd]

The test program above prints the values expected when run with the current master, but the whole program just doesn't quit after that. I see three RxNewThreadScheduler threads (2 core machine), all of them waiting for ThreadPoolExecutor.workQueue.take(). These threads aren't marked as daemon threads. See NewThreadScheduler L55.

Edit: Maybe they should be regular threads, but the executor running them should be allowed to timeout its single core thread since NewThreadScheduler will start a new pool anyway.

private static class EventLoopScheduler extends Scheduler {
    private final ExecutorService executor;

    private EventLoopScheduler() {
        ThreadPoolExecutor e = (ThreadPoolExecutor)Executors.newFixedThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "RxNewThreadScheduler-" + count.incrementAndGet());
            }
        });
        e.setKeepAliveTime(1, TimeUnit.SECONDS);
        e.allowCoreThreadTimeOut(true);
        executor = e;
    }
    ...
}