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


import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.Channel;
import com.yunfeiyun.agmp.common.framework.web.service.PortalService;
import com.yunfeiyun.agmp.common.core.domain.model.PortalInfo;
import com.yunfeiyun.agmp.common.framework.config.properties.RouterProperties;
import com.yunfeiyun.agmp.common.framework.web.service.ISysSyncService;
import com.yunfeiyun.agmp.common.utils.spring.SpringUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class MqAckReceiver implements ChannelAwareMessageListener {
    protected Logger log = LoggerFactory.getLogger(getClass());
    private PortalInfo portalInfo = SpringUtils.getBean(PortalService.class).getSsoInfo();
    @Autowired
    private RouterProperties routerProperties;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            MessageProperties properties = message.getMessageProperties();
            String topic = properties.getReceivedRoutingKey();
            int splitIdx = topic.lastIndexOf("/");
            topic = (splitIdx >= 0) ? topic.substring(splitIdx + 1) : topic;
            String messageBody = new String(message.getBody());
            log.info("通用方 deliveryTag={} topic={} , messageBody={}", deliveryTag, topic, messageBody);
            if(portalInfo.isSync()){
                MqMsg mqMsg =  (MqMsg) JSON.parseObject(messageBody, MqMsg.class);
                if (StringUtils.isBlank(mqMsg.getSeqId())) {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
                if(portalInfo.getAppName().equals(mqMsg.getSrc())){
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    return;
                }
                String msgSeq = mqMsg.getSeqId();
                String mqType = mqMsg.getMsgType();
                String mqCmd = mqMsg.getCmd();
                String mqSrc = mqMsg.getSrc();
                MqMsg   mqAck = SpringUtils.getBean(ISysSyncService.class).parseMsg(mqMsg);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("业务方 {} 接收到 {} 发送的消息 msgSeq={} mqType={}  mqCmd={}  mqBody={}  执行成功", portalInfo.getAppName(), mqSrc,  msgSeq,mqType, mqCmd , mqMsg.getData());
                if(routerProperties.getQueue().isAckEnabled()) ackMessage(mqAck);
            }
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
            MessageProperties properties = message.getMessageProperties();
            String messageBody = new String(message.getBody());
            String topic = properties.getReceivedRoutingKey();
            log.info("业务方 {}  topic={} , messageBody={} 发送的消息执行失败 重新放回队列尾部", portalInfo.getAppName(),topic, messageBody);
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }

    }

    public void ackMessage(MqMsg mqAck)  {
        try {
            if(mqAck !=null){
                MqMsg MqMsg = new MqMsg(mqAck.getMsgType(), mqAck.getCmd(), portalInfo.getAppName());
                MqMsg.setData( mqAck.getData() );
                SpringUtils.getBean(MqService.class).sendMsg(MqMsg);
            }
        }catch (Exception e){

        }
    }
}
