InterruptedException nel thread della cache RxJava durante il debug su Android

Question

A volte, quando eseguo il debug della mia app, ho riscontrato InterruptedException in RxCachedThreadScheduler-1. Ecco la traccia:

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) 

Ho una visualizzazione personalizzata in cui mi abbono al mio osservabile in questo modo:

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

Ecco il osservabile:

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

in cui eventSubject è un semplice PublishSubject. Sto usando RxAndroid 1.1.0 insieme a RxJava 1.1.0.

Qualcuno sa perché questo sta accadendo?

Answer 1

Non sono sicuro, perché succede, ma provo a fare questo:

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

Inoltre, penso che non è necessario share():

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

Io uso Subject come EventBus molto spesso, come ho descritto sopra. E non ho mai avuto problemi del genere.

P.S. In onDetachedFromWindow() sarebbe meglio se verificherai se l'iscrizione è stata annullata o meno. So che questo metodo chiamato nel thread principale, e l'accesso concorrente a questo Subscription è impossibile, ma penso che sia buono stile:

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