package com.bcxin.tenant.open.domains.components;

import com.bcxin.tenant.open.infrastructures.components.JsonProvider;
import com.bcxin.tenant.open.infrastructures.constants.KafkaConstants;
import com.bcxin.tenant.open.infrastructures.events.DomainEvent;
import com.bcxin.tenant.open.infrastructures.events.EventDispatcher;
import com.bcxin.tenant.open.infrastructures.events.IntegrationDomainEvent;
import com.bcxin.tenant.open.infrastructures.exceptions.BadTenantException;
import com.bcxin.tenant.open.infrastructures.utils.StringUtil;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class EventDispatcherImpl implements EventDispatcher {
    private final ApplicationEventPublisher eventPublisher;
    private final KafkaTemplate kafkaTemplate;
    private final JsonProvider jsonProvider;

    public EventDispatcherImpl(ApplicationEventPublisher eventPublisher, KafkaTemplate kafkaTemplate,
                               JsonProvider jsonProvider) {
        this.eventPublisher = eventPublisher;
        this.kafkaTemplate = kafkaTemplate;
        this.jsonProvider = jsonProvider;
    }

    @Override
    public void dispatch(DomainEvent event) {
        this.eventPublisher.publishEvent(event);
    }

    @Override
    public void dispatch(IntegrationDomainEvent event) {
        String key = event.getKey(jsonProvider);
        if (StringUtil.isEmpty(key)) {
            throw new BadTenantException("无效的key信息");
        }

        if (StringUtil.isEmpty(event.getEventName())) {
            throw new BadTenantException("无效的队列信息");
        }

        String value = event.getValue(jsonProvider);
        int partition = Math.abs(key.hashCode() % KafkaConstants.PARTITION_COUNT);


        this.kafkaTemplate.send(event.getEventName(), partition, key, value);
    }
}
