package cn.iocoder.yudao.module.iot.core.messagebus.core.rabbitmq;

import cn.hutool.core.util.TypeUtil;
import cn.iocoder.yudao.framework.common.util.json.JsonUtils;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
import jakarta.annotation.PreDestroy;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;

/**
 * 基于 RabbitMQ 的 {@link IotMessageBus} 实现类
 *
 * @author ywc
 */
@RequiredArgsConstructor
@Slf4j
public class IotRabbitMQMessageBus implements IotMessageBus {

    private static final String ROUTING_KEY = "#";

    private final RabbitTemplate rabbitTemplate;

    private final RabbitAdmin rabbitAdmin;

    @Getter
    private final List<IotMessageSubscriber<?>> subscribers = new ArrayList<>();

    private final List<SimpleMessageListenerContainer> containers = new ArrayList<>();

    @Override
    public void post(String topic, Object message) {
        rabbitTemplate.send(topic, ROUTING_KEY, MessageBuilder.withBody(JsonUtils.toJsonByte(message)).build());
        log.info("[post][topic({}) 发送消息({})]", topic, message);
    }

    @Override
    @SuppressWarnings("DataFlowIssue")
    public void register(IotMessageSubscriber<?> subscriber) {
        Type type = TypeUtil.getTypeArgument(subscriber.getClass(), 0);
        if (type == null) {
            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
        }

        // 1.1 声明交换机、队列和绑定关系
        Queue queue = new Queue(subscriber.getGroup(), true, false, false);
        rabbitAdmin.declareQueue(queue);
        TopicExchange exchange = new TopicExchange(subscriber.getTopic());
        rabbitAdmin.declareExchange(exchange);
        Binding binding = BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
        rabbitAdmin.declareBinding(binding);

        // 1.2 创建消费容器
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(rabbitTemplate.getConnectionFactory());
        container.setQueues(queue);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(10);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            try {
                subscriber.onMessage(JsonUtils.parseObject(message.getBody(), type));
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (Exception ex) {
                log.error("[onMessage][topic({}/{}) message({}) 消费者({}) 处理异常]",
                        subscriber.getTopic(), subscriber.getGroup(), message, subscriber.getClass().getName(), ex);
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            }
        });
        container.start();

        // 2. 保存消费者引用
        containers.add(container);
        subscribers.add(subscriber);
    }

    @PreDestroy
    public void destroy() {
        for (SimpleMessageListenerContainer container : containers) {
            try {
                container.stop();
                container.destroy();
                log.info("[destroy][关闭 RabbitMQ 消费者容器成功]");
            } catch (Exception e) {
                log.error("[destroy][关闭 RabbitMQ 消费者容器异常]", e);
            }
        }
    }

}