RxEventHub.java

RxEventHub.java

package io.github.zzdjk6;

import java.util.HashMap;

import rx.Observable;
import rx.subjects.PublishSubject;

@SuppressWarnings("WeakerAccess")
public final class RxEventHub {

    private final HashMap<String, PublishSubject> mSubjectMap = new HashMap<>();

    private final static RxEventHub s_sharedHub = new RxEventHub();

    public static RxEventHub getSharedHub() {
        return s_sharedHub;
    }

    @SuppressWarnings("unchecked")
    public final <T> Observable<T> event(RxEventProvider<T> provider) {
        PublishSubject subject = getEventSubject(provider);
        return subject.asObservable();
    }

    public final <T> void notify(RxEventProvider<T> provider, T data) {
        PublishSubject<T> subject = getEventSubject(provider);
        subject.onNext(data);
    }

    @SuppressWarnings("unchecked")
    private <T> PublishSubject<T> getEventSubject(RxEventProvider<T> provider) {
        String key = provider.typeKey();
        if (mSubjectMap.containsKey(key)) {
            PublishSubject subject = (PublishSubject<T>) mSubjectMap.get(key);
            if (subject != null)
                return subject;
        }

        PublishSubject subject = provider.publishSubject();
        mSubjectMap.put(key, subject);
        return subject;
    }
}

RxEventProvider.java

package io.github.zzdjk6;

import rx.subjects.PublishSubject;

@SuppressWarnings("WeakerAccess")
public class RxEventProvider<T> {

    public PublishSubject<T> publishSubject() {
        return PublishSubject.create();
    }

    public String typeKey() {
        return this.getClass().toString();
    }
}