Class EventBus

distributed-event-bus/src/main/java/EventBus.java:19–202

import java.util.function.Predicate;

@Singleton
public class EventBus {
    private final Map<String, List<Event>> topics;
    private final Map<String, Map<String, Integer>> eventIndexes;
    private final Map<String, ConcurrentSkipListMap<Long, String>> eventTimestamps;
    private final Map<String, Map<String, Subscription>> pullSubscriptions;
    private final Map<String, Map<String, Subscription>> pushSubscriptions;
    private final KeyedExecutor<String> eventExecutor;
    private final KeyedExecutor<String> broadcastExecutor;
    private final Timer timer;
    private EventBus deadLetterQueue;

    @Inject
    public EventBus(final KeyedExecutor<String> eventExecutor, final KeyedExecutor<String> broadcastExecutor, final Timer timer) {
        this.topics = new ConcurrentHashMap<>();
        this.eventIndexes = new ConcurrentHashMap<>();
        this.eventTimestamps = new ConcurrentHashMap<>();
        this.pullSubscriptions = new ConcurrentHashMap<>();
        this.pushSubscriptions = new ConcurrentHashMap<>();
        this.eventExecutor = eventExecutor;
        this.broadcastExecutor = broadcastExecutor;
        this.timer = timer;
    }

    public void setDeadLetterQueue(final EventBus deadLetterQueue) {
        this.deadLetterQueue = deadLetterQueue;
    }

    public CompletionStage<Void> publish(final String topic, final Event event) {
        return eventExecutor.getThreadFor(topic, publishToBus(topic, event));
    }

    private CompletionStage<Void> publishToBus(final String topic, final Event event) {
        if (eventIndexes.containsKey(topic) && eventIndexes.get(topic).containsKey(event.getId())) {
            return null;
        }
        topics.putIfAbsent(topic, new CopyOnWriteArrayList<>());
        eventIndexes.putIfAbsent(topic, new ConcurrentHashMap<>());
        eventIndexes.get(topic).put(event.getId(), topics.get(topic).size());
        eventTimestamps.putIfAbsent(topic, new ConcurrentSkipListMap<>());
        eventTimestamps.get(topic).put(timer.getCurrentTime(), event.getId());
        topics.get(topic).add(event);
        return notifyPushSubscribers(topic, event);
    }

    private CompletionStage<Void> notifyPushSubscribers(String topic, Event event) {
        if (!pushSubscriptions.containsKey(topic)) {
            return CompletableFuture.completedStage(null);
        }
        final var subscribersForTopic = pushSubscriptions.get(topic);
        final var notifications = subscribersForTopic.values()
                .stream()
                .filter(subscription -> subscription.getPrecondition().test(event))
                .map(subscription -> executeEventHandler(event, subscription))
                .toArray(CompletableFuture[]::new);
        return CompletableFuture.allOf(notifications);
    }