package com.fastbee.mqtt.utils;

import com.fastbee.common.core.mq.DeviceStatusBo;
import com.fastbee.common.enums.DeviceStatus;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.mqtt.model.ClientMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 服务器应答信息构建
 * @author gsb
 * @date 2022/10/7 14:17
 */
public class MqttMessageUtils {


    /**
     * 服务器确认连接应答消息 CONNACK
     */
    public static MqttConnAckMessage buildConntAckMessage(MqttConnectReturnCode code, boolean sessionPresent) {
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.CONNACK);
        MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(code, sessionPresent);
        return new MqttConnAckMessage(fixedHeader, variableHeader);
    }


    /**
     * 设备ping(心跳信息)应答 PINGRESP
     */
    public static MqttMessage buildPingResp() {
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PINGRESP);
        return new MqttMessage(fixedHeader);
    }

    /**
     * 取消订阅消息应答 UNSUBACK
     */
    public static MqttUnsubAckMessage buildUnsubAckMessage(MqttMessage message) {
        /*构建固定报文*/
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.UNSUBACK);
        return new MqttUnsubAckMessage(fixedHeader, getIdVariableHeader(message));
    }

    /**
     * 订阅确认应答 SUBACK
     */
    public static MqttSubAckMessage buildSubAckMessage(MqttMessage message) {
        /*构建固定报文*/
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.SUBACK);
        /*构建可变报文*/
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) message;

        /*获取订阅topic的Qos*/
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(MqttTopicSubscription::topicName).collect(Collectors.toSet());
        List<Integer> grantedQos = new ArrayList<>(topics.size());
        for (int i = 0; i < topics.size(); i++) {
            grantedQos.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        /*负载*/
        MqttSubAckPayload payload = new MqttSubAckPayload(grantedQos);
        return new MqttSubAckMessage(fixedHeader, getIdVariableHeader(message), payload);
    }

    /**
     * 构建推送应答消息 PUBLISH
     */
    public static MqttPublishMessage buildPublishMessage(ClientMessage msg, int packageId) {
        /*报文固定头*/
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, msg.isDup(), msg.getQos(), false, 0);
        /*报文可变头*/
        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(msg.getTopicName(), packageId);
        /*负载*/
        ByteBuf payload = msg.getPayload() == null ? Unpooled.EMPTY_BUFFER : Unpooled.wrappedBuffer(msg.getPayload());
        /*完整报文，固定头+可变头+payload*/
        return new MqttPublishMessage(fixedHeader, variableHeader, payload);
    }

    /**
     * Qos1 收到发布消息确认 无负载 PUBACK
     */
    public static MqttPubAckMessage buildAckMessage(MqttMessage message) {
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBACK);
        return new MqttPubAckMessage(fixedHeader, getIdVariableHeader(message));
    }

    /**
     * Qos2 发到消息收到 无负载 PUBREC
     */
    public static MqttMessage buildPubRecMessage(MqttMessage message){
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBREC);
        return new MqttMessage(fixedHeader, getIdVariableHeader(message));
    }

    /**
     * Qos2 发布消息释放 PUBREL
     */
    public static MqttMessage buildPubRelMessage(MqttMessage message){
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBREL);
        return new MqttMessage(fixedHeader, getIdVariableHeader(message));

    }

    /**
     * Qos2 发布消息完成 PUBCOMP
     */
    public static MqttMessage buildPubCompMessage(MqttMessage message){
        MqttFixedHeader fixedHeader = buildFixedHeader(MqttMessageType.PUBCOMP);
        return new MqttMessage(fixedHeader, getIdVariableHeader(message));
    }

    /**
     * 固定头定制
     */
    public static MqttFixedHeader buildFixedHeader(MqttMessageType messageType) {
        return new MqttFixedHeader(messageType, false, MqttQoS.AT_MOST_ONCE, false, 0);
    }

    /**
     * 构造MqttMessageIdVariableHeader
     */
    public static MqttMessageIdVariableHeader getIdVariableHeader(MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader idVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        return MqttMessageIdVariableHeader.from(idVariableHeader.messageId());
    }

    /*构造返回MQ的设备状态model*/
    public static DeviceStatusBo buildStatusMsg(ChannelHandlerContext ctx, String clientId,DeviceStatus status,String ip){
        return DeviceStatusBo.builder()
                .serialNumber(clientId)
                .status(status)
                .ip(ip)
                .hostName(ip)
                .timestamp(DateUtils.getNowDate()).build();
    }
}