InterruptedException in RxJava cache thread when debugging on Android

Question

Sometimes, when I debug my application, I encounter an InterruptedException in RxCachedThreadScheduler-1. Here's the trace:

Fatal Exception: java.lang.InterruptedException at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:1991) at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2025) at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1048) at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:776) at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1035) 

I have a custom view in which I subscribe to my observables as follows:

 @Override protected void onAttachedToWindow() { super.onAttachedToWindow(); sub = FollowHandler.getInstance().getObservable() .filter(new Func1<FollowEvent, Boolean>() { @Override public Boolean call(FollowEvent followEvent) { if(followEvent == null || followEvent.user == null || user == null) return false; return followEvent.user.id == user.id; } }) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Observer<FollowEvent>() { @Override public void onCompleted() {} @Override public void onError(Throwable e) {} @Override public void onNext(FollowEvent followEvent) { reactToThisNiceEvent(followEvent); } }); } @Override protected void onDetachedFromWindow() { super.onDetachedFromWindow(); if(sub != null) sub.unsubscribe(); } 

Here's the observable:

 eventSubject.asObservable() .observeOn(Schedulers.io()) .doOnNext(new Action1<FollowEvent>() { @Override public void call(FollowEvent followEvent) { if(followEvent != null) doSomethingNice(followEvent); } }) .share(); 

in which eventSubject is a simple PublishSubject. I am using RxAndroid 1.1.0 along with RxJava 1.1.0.

Does anyone know why this is happening?

Answer 1

I'm not sure why this is happening, but try to do this:

 sub = FollowHandler.getInstance().getObservable() .filter(...) .subscribeOn(Schedulers.io()) // <<<<<<<<<< .observeOn(AndroidSchedulers.mainThread()) .subscribe(...); 

Also, I think you do not need share() :

 eventSubject.asObservable() .doOnNext(...) .subscribeOn(Schedulers.io()) // <<<<< subscribeOn instead of observeOn, but actually, you don't need it here... .share(); // <<<<< remove it 

I use Subject as eventbus very often, as described above. And I never had such problems.

PS In onDetachedFromWindow() it would be better if you check the subscription for the subscription or not. I know that this method is called in the main thread, and simultaneous access to this Subscription not possible, but I think this is a good style:

 if(sub != null && !sub.isUnsubscribed()) sub.unsubscribe();