Rxjava Event Bus erneut abonnieren auf Wiederaufnahme funktioniert nicht

Question

Ich bin neu zu RxJava und ich versuche RxJava als Event-Bus zu verwenden. Ich verwende RxJava Beispiel von Kaushik Gopal (https://github.com/kaushikgopal/RxJava-Android-Samples/blob/master/app/src/main/java/com/morihacky/android/rxjava/rxbus/RxBus.java).

Hier ist mein RxBus Code

public class RxBus { 

private static RxBus _instance; 
private final Subject<Object, Object> _bus; 

public static synchronized RxBus getInstance(){ 
    if(_instance == null){ 
     _instance = new RxBus(); 
    } 
    return _instance; 
} 

private RxBus() { 
    _bus = new SerializedSubject<>(PublishSubject.create()); 
} 

public void send(Object o) { 
    _bus.onNext(o); 
} 

public Observable<Object> toObserverable() { 
    return _bus; 
} 

public boolean hasObservers() { 
    return _bus.hasObservers(); 
} 
} 

Ich bin auf einem anderen Thread zu veröffentlichen, aber ich bin auf einem UI-Thread abonnieren möchte. Hier ist mein Abonnement-Code

public abstract class BaseActivity extends AppCompatActivity { 

     protected CompositeSubscription _subscriptions; 

     @Override 
     protected void onCreate(Bundle savedInstanceState) { 
      ... 
      _subscriptions = new CompositeSubscription(); 
      ... 
     } 

     @Override 
     protected void onResume() { 

      .... 
      _subscriptions.add(RxBus.getInstance().toObserverable() 
        .observeOn(AndroidSchedulers.mainThread()) 
        .subscribe(new Action1<Object>() { 
         @Override 
         public void call(final Object event) { 
          //Do something here 
         } 
        })); 

      .....   
     } 

     @Override 
     protected void onPause() { 

      ...... 
      _subscriptions.unsubscribe(); 
    } 
} 

Wenn ich zuerst die Aktivität lade ich die Ereignisse erhalten. Aber wenn ich die Aktivität wieder schließe und öffne (trigger onPause und onResume), höre ich auf, die Ereignisse zu bekommen. Fehle ich hier etwas? Ist das der richtige Weg, RxJava für den Event-Bus zu nutzen? Ich poste wieder auf einem anderen Thread, aber ich abonniere auf einem UI-Thread. Danke

Answer 1

Ich erkannte das Problem. Musste _subscriptions.clear() statt _subscriptions.unsubscribe() aufrufen;