package com.skr.mdm.rabbitmq;

import cn.hutool.json.JSONUtil;
import com.rabbitmq.client.Channel;
import com.skr.mdm.bean.so.RabbitMqSO;
import com.skr.mdm.config.MdmWebsocketChannelFactory;
import com.skr.mdm.constant.CmdTypeConstants;
import com.skr.mdm.constant.RabbitmqBindingKeyConstants;
import com.skr.mdm.constant.RabbitmqExchangeConstants;
import com.skr.mdm.netty.WebsocketEncoder;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * @author nfq
 * @date 2020/6/16 16:22
 */
@Slf4j
@Component
@RabbitListener(bindings = {
        @QueueBinding(
                value = @Queue,
                exchange = @Exchange(value = RabbitmqExchangeConstants.TOPIC_EXCHANGE, type = "topic"),
                key = {RabbitmqBindingKeyConstants.WEBSOCKET_BINDING_KEY})})
public class RabbitmqConsumer {

    @Value("${mdm.encrypt}")
    private String mdmEncrypt;

    @RabbitHandler
    public void handle(RabbitMqSO rabbitMqSO, @Headers Map<String, Object> headers, Channel channel) {
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            if (rabbitMqSO == null || CollectionUtils.isEmpty(rabbitMqSO.getChannelIdList())) {
                return;
            }
//            log.info("MQ收到下发消息 channelIdList：{} 下发内容：{}", JSONUtil.toJsonStr(rabbitMqSO.getChannelIdList()), rabbitMqSO.getMessage());
            List<String> channelIdList = rabbitMqSO.getChannelIdList();
            channelIdList.forEach(channelId -> {
                ChannelHandlerContext ctx = MdmWebsocketChannelFactory.getChannel(channelId);
                if (ctx != null) {
//                    log.info("channelId：{}存在并推送", channelId);
                    ByteBuf byteBuf = WebsocketEncoder.encoder(CmdTypeConstants.ISSUE, rabbitMqSO.getMessage(), mdmEncrypt);
                    // 使用EventLoop的任务调度
                    ctx.channel().eventLoop().execute(() -> {
                                ctx.channel().writeAndFlush(new BinaryWebSocketFrame(byteBuf));
                            }
                    );
//                    log.info("channelId：{}推送成功", channelId);
                } else {
//                    log.info("channelId：{}不存在", channelId);
                }
            });
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            try {
                log.error("rabbitmq接收下发信息异常：" + JSONUtil.toJsonStr(rabbitMqSO), e);
                channel.basicAck(deliveryTag, false);
            } catch (Exception ioException) {
                log.error("rabbitmq消息确认异常：" + JSONUtil.toJsonStr(rabbitMqSO), e);
            }
        }
    }

}
