Description
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?