package com.yunfeiyun.agmp.common.framework.config;

import com.yunfeiyun.agmp.common.config.SystemConfig;
import com.yunfeiyun.agmp.common.constant.AmqpConstant;
import com.yunfeiyun.agmp.common.framework.mq.MqAckReceiver;
import com.yunfeiyun.agmp.common.framework.config.properties.RouterProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.annotation.Order;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @author   杨晓辉
 */
@Order(0)
@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "enabled", havingValue = "true")
@Configuration
@Import(TopicRabbitConfig.class)
public class MqConfiguration {

    protected Logger log = LoggerFactory.getLogger(getClass());

    @Bean
    public RabbitAdmin rabbitAdmin() {
        return new RabbitAdmin(connectionFactory);
    }

    @Autowired
    private SystemConfig systemConfig;

    @Resource
    private CachingConnectionFactory connectionFactory;

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    private RabbitAdmin rabbitAdmin;

    @Autowired
    private RouterProperties routerProperties;

    @PostConstruct
    public void initRabbitTemplate(){
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        connectionFactory.setPublisherReturns(true);
        rabbitTemplate.setMandatory(false);
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String s) {
                log.info("消息进入到broker，是否到交换机:correlationData({}),ack({}),cause({})", correlationData, ack, s);
                if(ack){
                    //mailSendLogService.updateStatusByMsgId(1,correlationData.getId());
                    log.info("消息到达交换机 发送成功，更新状态为1");
                }
            }
        });// 消息未进入队列时触发回调
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returnedMessage) {
                log.info("消息未进入队列:exchange({}),route({}),replyCode({}),replyText({}),message:{}",returnedMessage.getExchange(), returnedMessage.getRoutingKey(), returnedMessage.getReplyCode(), returnedMessage.getReplyText(), returnedMessage.getMessage());
            }
        });


    }


    /**
     * 消息接收处理类
     */
    @Resource
    private MqAckReceiver mqAckReceiver;

    @Bean
    @ConditionalOnProperty(name = "portal.syncEnabled", matchIfMissing = false)
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        if(routerProperties.getQueue() != null){
                List<String> router = routerProperties.getQueue().getConsumeRouter();
                List<String> routerList = router.stream().map(queueName ->{
                    Properties properties = rabbitAdmin.getQueueProperties(queueName);
                    if(properties == null){
                        log.info("业务方={} 监听  队列={} 不存在创建",systemConfig.getName() , queueName);
                        Queue queue = new Queue(queueName, true, false, false, null);
                        FanoutExchange fanoutExchange = new FanoutExchange(AmqpConstant.FanoutExchangeName);
                        rabbitAdmin.declareQueue(queue);
                        rabbitAdmin.declareExchange(fanoutExchange);
                        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(fanoutExchange));
                    }
                    return "" + queueName;
                }).collect(Collectors.toList());
                container.setQueueNames(routerList.toArray(new String[routerList.size()]));
                log.info("业务方={}  队列={}  消费者队列监听成功",systemConfig.getName() , routerList);
        }else{
            log.error("您没有配置消费者队列");
        }
        container.setExposeListenerChannel(true);
        container.setPrefetchCount(1);//设置每个消费者获取的最大的消息数量
        container.setConcurrentConsumers(1);//消费者个数
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);//设置确认模式为手工确认
        container.setMessageListener(mqAckReceiver);


        return container;
    }





}
