NotificationLoop.java

package fuck.josh;

import java.util.HashMap;
import java.util.Map;

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

public class NotificationLoop {
    
    private final Map<Class<?>, PublishSubject<?>> containers = new HashMap<>();
    
    private <R> PublishSubject<R> getOrCreatePS(Class<R> tClass) {
        PublishSubject<?> publishSubject = containers.get(tClass);
        if (publishSubject == null) {
            publishSubject = PublishSubject.create();
            containers.put(tClass, publishSubject);
        }
        return (PublishSubject<R>) publishSubject;
    }
    
    public <R> Observable<R> observeEvent(Class<R> clas) {
        PublishSubject<R> publishSubject = getOrCreatePS(clas);
        return publishSubject.asObservable();
    }
    
    public <R> void callEvent(R r) {
        containers.entrySet().forEach(entry -> {
            if (entry.getKey().isAssignableFrom(r.getClass())) {
                PublishSubject<R> tPublishSubject = (PublishSubject<R>) entry.getValue();
                tPublishSubject.onNext(r);
            }
        });
    }
}