public class RxBus {
    private final Map<Class, PublishSubject> subjects = new HashMap<>();

    @SuppressWarnings("unchecked")
    public synchronized <T> void emit(final T value) {
        for (Class clazz = value.getClass(); clazz != null; clazz = clazz.getSuperclass()) {
            onGetOrCreateStream(clazz).onNext(value);
        }
    }

    @SuppressWarnings("unchecked")
    public synchronized <T> Observable<T> stream(final Class<T> clazz) {
        return onGetOrCreateStream(clazz);
    }

    @SuppressWarnings("unchecked")
    private <T> PublishSubject<T> onGetOrCreateStream(final Class<T> clazz) {
        if (!subjects.containsKey(clazz)) {
            subjects.put(clazz, PublishSubject.<T>create());
        }
        return subjects.get(clazz);
    }
}