Rxjava 2 Bus

build.gradle

dependencies {
    compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
}

RxBus.java

/**
 * for send object between fragment, activity, service...
 * REMEMBER: dispose if not in use to avoid leak memory
 */

public class RxBus<T> {
    private static final String PREFIX = "$^%)@";
    private static RxBus instance;
    private PublishSubject<BusMessage<T>> subject = PublishSubject.create();

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

    /**
     * send message with chanel to receive
     *
     * @param chanel  to filter between multiple object sent
     * @param message
     * @see #getMessages(String)
     */
    public void sendMessage(String chanel, T message) {
        subject.onNext(new BusMessage<T>(chanel, message));
    }

    /**
     * @param chanel to filter between multiple object sent
     * @return Observable object with chanel filter
     * @see #sendMessage(String, Object)
     */
    public Observable<T> getMessages(String chanel) {
        return subject
                .filter(tBusMessage -> tBusMessage.chanel == chanel) // filter by chanel
                .map(tBusMessage -> tBusMessage.value); // custom output (remove chanel+prefix string)
    }

    /**
     * message object to send
     * @param <T>
     */
    class BusMessage<T> {
        String chanel;
        T value;

        public BusMessage(String key, T value) {
            this.chanel = key;
            this.value = value;
        }
    }
}

usage

    // unsubscribe if not in use
    protected void onPause() {
        super.onPause();
        unsubscriber.dispose();
    }

    
    protected void onResume() {
        super.onResume();
        unsubscriber = RxBus.getInstance()
                .getMessages(ContactAdapter.ON_CONTACT_CLICK)
                .subscribe(onContactClick());
    }
    private Consumer<Contact> onContactClick() {
        return contact -> {
            LogUtils.print(contact.getName()+","+contact.getNumbers());
        };
    }
    
    
    /// send message from adapter
    itemHolder.itemView.setOnClickListener(view -> RxBus.getInstance().sendMessage(ON_CONTACT_CLICK, contact));
    
    /// you can send any type of object: String, Integer, Object class... => create Consumer<YOUR_TYPE> on .subscribe