package de.accso.flexinale.common.infrastructure.eventbus;

import de.accso.flexinale.common.api.eventbus.EventBus;
import de.accso.flexinale.common.api.eventbus.EventBusFactory;
import de.accso.flexinale.common.api.event.Event;

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

@SuppressWarnings("unused")
public final class KafkaAsyncEventBusFactory implements EventBusFactory {
    private final Map <    Class<? extends Event>,
                        EventBus<? extends Event>  > eventTypeToEventBusMap = new HashMap<>();

    private final KafkaAvailabilityChecker kafkaAvailabilityChecker;

    public KafkaAsyncEventBusFactory(final KafkaAvailabilityChecker kafkaAvailabilityChecker) {
        this.kafkaAvailabilityChecker = kafkaAvailabilityChecker;
    }

    @Override
    public <E extends Event> EventBus<E> createOrGetEventBusFor(final Class<E> eventType) {
        EventBus<? extends Event> eventBus = eventTypeToEventBusMap.get(eventType);
        if (eventBus == null) {
            eventBus = new KafkaAsyncEventBus<>(eventType, kafkaAvailabilityChecker);
            eventTypeToEventBusMap.put(eventType, eventBus);
        }
        return (EventBus<E>) eventBus;
    }
}
