RxBus 封装及使用

发送消息

RxBus.getDefault().post(UnReadCount(from = "MessageIMFragment",content = unreadNum.toString()))

接收消息

RxBus.getDefault().toFlowable(UnReadCount::class.java).subscribe { it ->
                if (it.type == UNREADE_P2P && it.content.toInt() > 0) {
                    unread.text = it.content
                    unread.visibility = View.VISIBLE
                }else{
                    unread.visibility = View.GONE
                }
            }

数据类

data class UnReadCount(
        val type: String = UNREADE_P2P,
        val from: String = "",
        val content: String = "0"
)

封装

/**
 * <pre>
 *     author : jake
 *     time   : 2018/03/23
 *     function   :  RxBus 封装
 *     version: 1.0
 * </pre>
 */
public class RxBus {
    private final FlowableProcessor<Object> mBus;
    private static volatile RxBus sRxBus = null;

    private RxBus() {
        //调用toSerialized()方法，保证线程安全
        mBus = PublishProcessor.create().toSerialized();
    }

    public static synchronized RxBus getDefault() {
        if (sRxBus == null) {
            synchronized (RxBus.class) {
                if (sRxBus == null) {
                    sRxBus = new RxBus();
                }
            }
        }
        return sRxBus;
    }


    /**
     * 发送消息
     *
     * @param o
     */
    public void post(Object o) {
        new SerializedSubscriber<>(mBus).onNext(o);
    }

    /**
     * 确定接收消息的类型
     *
     * @param aClass
     * @param <T>
     * @return
     */
    public <T> Flowable<T> toFlowable(Class<T> aClass) {
        return mBus.ofType(aClass);
    }

    /**
     * 判断是否有订阅者
     *
     * @return
     */
    public boolean hasSubscribers() {
        return mBus.hasSubscribers();
    }

}