EventBus

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

// Concrete EventBus implementation
public class SimpleEventBus<T> implements MyEventBus<T> {

    private final ConcurrentHashMap<String, List<MyEventListener<T>>> listenersMap = new ConcurrentHashMap<>();

    // Publish an event to all listeners of a source
    public void push(String source, T event) {
        List<MyEventListener<T>> listeners = listenersMap.get(source);
        if (listeners != null && !listeners.isEmpty()) {
            for (MyEventListener<T> listener : listeners) {
                try {
                    listener.onEvent(source, event);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    // Get number of subscribers for a source
    public int getSubscriberCount(String source) {
        List<MyEventListener<T>> listeners = listenersMap.get(source);
        return (listeners == null) ? 0 : listeners.size();
    }

    @Override
    public void subscribe(String source, MyEventListener<T> listener) {
        listenersMap.computeIfAbsent(source, key -> new CopyOnWriteArrayList<>()).add(listener);
    }

    @Override
    public void unsubscribe(String source, MyEventListener<T> listener) {
        listenersMap.computeIfPresent(source, (key, list) -> {
            list.remove(listener);
            return list.isEmpty() ? null : list;
        });
    }
}

class MessageListener implements MyEventListener<Message> {
    private final String name;

    public MessageListener(String name) { this.name = name; }

    @Override
    public void onEvent(String source, Message event) {
        System.out.println(name + " received from [" + source + "]: " + event.getText());
    }
}