RxJava as event bus in Android project - remove event from the bus

Question

I used to work with EventBus before, which was easy to use and easy to understand. This time, however, I would like to try out RxJava for event bus-like communication, however it is not very clear how to remove events from RxJava or, better to say, how it should be designed properly to have similar behavior as EventBus has when I call removeStickyEvent?

In RxJava I can use BehaviorSubject to reply last even when I subscribe to this observable, but what should I do when this event is handled? What if I do not want to have this event replayed again?

For instance, one fragment fires an event and then finishes. Another fragment listens to this event and handles it. Then, if this app fires that "another" activity again from different circumstances, then it will subscribe to the same BehaviorSubject again and will handle that stale event again, which is not what I would like to achieve.

I used this project as a reference https://github.com/marwinxxii/AndroidRxSamples/blob/master/app/src/main/java/com/github/marwinxxii/rxsamples/EventBusSampleActivity.java

Excuse me for a non-answer, but.. it might be easier for you to learn the idea behind RxJava and it's idiomatic use and patterns, instead of trying to re-implement a different pattern that just might make no sense in RxJava. — Marcin Koziński

Bartek Lipinski Bartek Lipinski · Accepted Answer · 2016-06-07T07:44:02

As long as you do not plan to allow your events to be null, I think that can be achieved pretty easily.

Exactly as you said, you could use BehaviorSubject to propagate sticky events, and when you want to removeStickyEvent from the bus, you can just emit a null object (to "flush" the subject).

Something like this (from the top of my head - not tested, without generics, just a simple Object-event based example):

public class RxEventBus {

    PublishSubject<Object> eventsSubject = PublishSubject.create();
    BehaviorSubject<Object> stickyEventsSubject = BehaviorSubject.create();

    public RxEventBus() {
    }

    public Observable<Object> asObservable() {
        return eventsSubject;
    }

    public Observable<Object> asStickyObservable() {
        return stickyEventsSubject.filter(new Func1<Object, Boolean>() {
            @Override
            public Boolean call(Object o) {
                return o != null;
            }
        });
    }

    public void postEvent(@NonNull Object event) {
        eventsSubject.onNext(event);
    }

    public void postStickyEvent(@NonNull Object stickyEvent) {
        stickyEventsSubject.onNext(stickyEvent);
    }

    public void removeStickyEvent(){
        stickyEventsSubject.onNext(null);
    }
}

RxJava as event bus in Android project - remove event from the bus

1 Answers