kaushikgopal

// don't start emitting items just yet by turning the observable to a connected one
ConnectableObservable<Object> tapEventEmitter = _rxBus.toObserverable().publish();

tapEventEmitter.publish((Func1) (stream) -> {

    // inside `publish`, "stream" is truly multicasted

    // applying the same technique for getting a debounced buffer sequence
    return stream.buffer(stream.debounce(1, TimeUnit.SECONDS));

kaushikgopal

Observable<Object> tapEventEmitter = _rxBus.toObserverable().share();
Observable<Object> debouncedEventEmitter = tapEventEmitter.debounce(1, TimeUnit.SECONDS);
tapEventEmitter.buffer(debouncedEventEmitter)

kaushikgopal

Observable<Object> tapEventEmitter = _rxBus.toObserverable().share();
// which is really the same as:
Observable<Object> tapEventEmitter = _rxBus.toObserverable().publish().refcount();

kaushikgopal

    Observable<Object> tapEventEmitter = _rxBus.toObserverable().share();
    Observable<Object> debouncedEventEmitter = tapEventEmitter.debounce(1, TimeUnit.SECONDS);
    Observable<List<Object>> debouncedBufferEmitter = tapEventEmitter.buffer(debouncedEventEmitter);

    debouncedBufferEmitter.buffer(debouncedEventEmitter)
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Action1<List<Object>>() {
          @Override
          public void call(List<Object> taps) {
            _showTapCount(taps.size());

kaushikgopal

// note that it is important to subscribe to the exact same _rxBus instance that was used to post the events
_rxBus.toObserverable()
    .subscribe(new Action1<Object>() {
      @Override
      public void call(Object event) {

        if(event instanceof TapEvent) {
          _showTapText();

        }else if(event instanceof SomeOtherEvent) {

kaushikgopal

@OnClick(R.id.btn_demo_rxbus_tap)
public void onTapButtonClicked() {

    _rxBus.send(new TapEvent());
}

kaushikgopal

// this is the middleman object
public class RxBus {

  private final Subject<Object, Object> _bus = new SerializedSubject<>(PublishSubject.create());

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

  public Observable<Object> toObserverable() {

kaushikgopal

// the below code is inside a TextWatcher
// which implements the onTextChanged method
// I've simplified it to only highlight the parts we're
// interested in

private long lastChange = 0;

@Override
public void onTextChanged(final CharSequence chars,

kaushikgopal

public class TestActivity extends Activity {

    // ...
    // all standard stuff

    @Override
    public void onCreate(Bundle savedInstanceState) {

        // ...
        // all standard stuff

kaushikgopal

// Version 1
public class IAmAThread extends Thread {
    public IAmAThread() {
        super("IAmAThread");
    }

    @Override
    public void run() {
        // your code (sequence of instructions)