RxBus for RxJava 1 and RxJava 2

RxBus1.java

import rx.Observable;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;

/**
 * @author <a href="mailto:jaredsburrows@gmail.com">Jared Burrows</a>
 */
public final class RxBus {
    private final Subject<Object, Object> bus = new SerializedSubject<>(PublishSubject.create());

    public void send(final Object event) {
        bus.onNext(event);
    }

    public Observable<Object> toObservable() {
        return bus;
    }

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

RxBus1RxRelay1.java

import com.jakewharton.rxrelay.PublishRelay;
import com.jakewharton.rxrelay.Relay;

import rx.Observable;

/**
 * @author <a href="mailto:jaredsburrows@gmail.com">Jared Burrows</a>
 */
public final class RxBus {
    private final Relay<Object, Object> bus = PublishRelay.create().toSerialized();

    public void send(final Object event) {
        bus.call(event);
    }

    public Observable<Object> toObservable() {
        return bus;
    }

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

RxBus2.java

import io.reactivex.Observable;
import io.reactivex.subjects.PublishSubject;

/**
 * @author <a href="mailto:jaredsburrows@gmail.com">Jared Burrows</a>
 */
public final class RxBus {
    private final PublishSubject<Object> bus = PublishSubject.create();

    public void send(final Object event) {
        bus.onNext(event);
    }

    public Observable<Object> toObservable() {
        return bus;
    }

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

RxBus2RxRelay2.java

import com.jakewharton.rxrelay2.PublishRelay;
import com.jakewharton.rxrelay2.Relay;
import io.reactivex.Observable;

/**
 * @author <a href="mailto:jaredsburrows@gmail.com">Jared Burrows</a>
 */
public final class RxBus {
  private final Relay<Object> bus = PublishRelay.create().toSerialized();

  public void send(Object event) {
    bus.accept(event);
  }

  public Observable<Object> toObservable() {
    return bus;
  }

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

For my own bus I mostly subscribe to one specific event as that turned out to make clearer separation:

public <T> Observable<T> event(Class<T> klazz) {
    return mBus.ofType(klazz);
}

I think is better to use RxRelay from Jake Wharton... because if onError occurs, the bus is dead.

@PaulWoitaschek That is great for filtering events to listen for.

@igoticecream Yeah. I'll update the gists with https://github.com/JakeWharton/RxRelay.

A activity send msg ， B activity can receive msg？

@xwc520 You can do that with the RxBus.

How do you unsubscribe?

ReactiveX/RxJava#5575