package com.jay.monitor.data.client.utils;

import com.jay.monitor.data.core.enums.MQInvokeType;
import com.jay.monitor.data.core.enums.MQTypeEnum;
import com.jay.monitor.data.core.model.serializable.MQDataDTO;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Date;

/**
 * @Module 工具
 * @Description 将一些通用的监控数据方法进行抽取
 * @Author liukaixiong
 * @Date 2020/10/30 17:02
 */
public class MonitorUtils {

    public static MQDataDTO builderMqDataDTO(String topic, String key, String partitionName, String msgId, String value, Integer status, Long requestTime) {
        return builderMqDataDTO(topic, key, partitionName, msgId, value, status, requestTime, null);
    }

    /**
     * 构建队列数据
     *
     * @param topic         主题
     * @param key           key
     * @param partitionName 分区
     * @param msgId         消息编号
     * @param value         值
     * @param status        状态
     * @param requestTime   请求时长
     * @param e             错误信息
     * @return
     */
    public static MQDataDTO builderMqDataDTO(String topic, String key, String partitionName, String msgId, String value, Integer status, Long requestTime, Throwable e) {
        MQDataDTO mq = new MQDataDTO();
        mq.setTopic(topic);
        mq.setMsgId(msgId);
        mq.setCreateDate(new Date());
        mq.setMqType(MQTypeEnum.kafka.name());
        mq.setPartitionName(partitionName);
        mq.setJsonContent(value);
        mq.setGroupKeyName(key);
        mq.setStatus(status);
        mq.setRequestTime(requestTime);
        mq.setInvokeType(MQInvokeType.producer);

        if (e != null) {
            StringWriter writer = new StringWriter(2048);
            e.printStackTrace(new PrintWriter(writer));
            mq.setErrorMessage(e.getMessage());
            mq.setErrorStackTrace(writer.toString());
        }

        return mq;
    }


    public static MQDataDTO builderConsumerRMQDataDTO(String topic, String key, String partitionName, Object msgId, String value, Integer status, Long requestTime) {
        return builderConsumerRMQDataDTO(topic, key, partitionName, msgId, value, status, requestTime, null);
    }

    /**
     * 构建RMQ的消费者消息
     *
     * @param topic         topic
     * @param key           key
     * @param partitionName 分区号
     * @param msgId         消息编号
     * @param value         值
     * @param status        状态
     * @param requestTime   请求时长
     * @param e             错误信息
     * @return
     */
    public static MQDataDTO builderConsumerRMQDataDTO(String topic, String key, String partitionName, Object msgId, String value, Integer status, Long requestTime, Throwable e) {
        MQDataDTO mqDataDTO = builderMqDataDTO(topic, key, partitionName, String.valueOf(msgId), value, status, requestTime, e);
        mqDataDTO.setInvokeType(MQInvokeType.consumer);
        mqDataDTO.setMqType(MQTypeEnum.rocketMQ.name());
        return mqDataDTO;
    }


    public static MQDataDTO builderProducerMqDataDTO(String topic, String key, String partitionName, Object msgId, String value, Integer status) {
        MQDataDTO mqDataDTO = builderMqDataDTO(topic, key, partitionName, String.valueOf(msgId), value, status, null);
        mqDataDTO.setInvokeType(MQInvokeType.producer);
        return mqDataDTO;
    }

    /**
     * 构建生产者监控数据
     *
     * @param topic         topic
     * @param key           key
     * @param partitionName kafka/分区 -> RMQ/Tag
     * @param msgId         消息编号
     * @param value         消息内容
     * @param status        成功或者失败
     * @param e             异常信息
     * @return
     */
    public static MQDataDTO builderProducerMqDataDTO(String topic, String key, String partitionName, Object msgId, String value, Integer status, Throwable e) {
        MQDataDTO mqDataDTO = builderMqDataDTO(topic, key, partitionName, String.valueOf(msgId), value, status, null, e);
        mqDataDTO.setInvokeType(MQInvokeType.producer);
        return mqDataDTO;
    }

    public static MQDataDTO builderConsumerMqDataDTO(String topic, String key, String partitionName, Object msgId, String value, Integer status, Long requestTime) {
        return builderConsumerMqDataDTO(topic, key, partitionName, msgId, value, status, requestTime, null);
    }

    /**
     * 构建消费者监控数据
     *
     * @param topic         topic
     * @param key           key
     * @param partitionName kafka/分区 -> RMQ/Tag
     * @param msgId         消息编号
     * @param value         消息内容
     * @param status        成功或者失败
     * @param e             异常信息
     * @return
     */
    public static MQDataDTO builderConsumerMqDataDTO(String topic, String key, String partitionName, Object msgId, String value, Integer status, Long requestTime, Throwable e) {
        MQDataDTO mqDataDTO = builderMqDataDTO(topic, key, partitionName, String.valueOf(msgId), value, status, requestTime, e);
        mqDataDTO.setInvokeType(MQInvokeType.consumer);
        return mqDataDTO;
    }


}
