package com.sohu.tv.mq.cloud.web.controller;

import com.sohu.tv.mq.cloud.bo.*;
import com.sohu.tv.mq.cloud.bo.Audit.TypeEnum;
import com.sohu.tv.mq.cloud.common.util.WebUtil;
import com.sohu.tv.mq.cloud.service.*;
import com.sohu.tv.mq.cloud.util.*;
import com.sohu.tv.mq.cloud.web.controller.param.*;
import com.sohu.tv.mq.cloud.web.vo.ConsumerProgressVO;
import com.sohu.tv.mq.cloud.web.vo.ConsumerRunningInfoVO;
import com.sohu.tv.mq.cloud.web.vo.QueueOwnerVO;
import com.sohu.tv.mq.cloud.web.vo.UserInfo;
import com.sohu.tv.mq.metric.StackTraceMetric;
import com.sohu.tv.mq.util.CommonUtil;
import com.sohu.tv.mq.util.Constant;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.util.HtmlUtils;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.validation.Valid;
import java.util.*;
import java.util.stream.Collectors;

import static com.sohu.tv.mq.cloud.bo.Audit.TypeEnum.UPDATE_HTTP_CONSUMER_CONFIG;

/**
 * 消费者接口
 * 
 * @Description:
 * @author yongfeigao
 * @date 2018年6月12日
 */
@Controller
@RequestMapping("/consumer")
public class ConsumerController extends ViewController {

    @Autowired
    private ConsumerService consumerService;

    @Autowired
    private UserService userService;

    @Autowired
    private UserConsumerService userConsumerService;

    @Autowired
    private AuditService auditService;

    @Autowired
    private AlertService alertService;

    @Autowired
    private TopicService topicService;

    @Autowired
    private VerifyDataService verifyDataService;

    @Autowired
    private ClusterService clusterService;

    @Autowired
    private ConsumerConfigService consumerConfigService;

    @Autowired
    private MQCloudConfigHelper mqCloudConfigHelper;

    @Autowired
    private ClientVersionService clientVersionService;

    @Autowired
    private UserFootprintService userFootprintService;

    @Autowired
    private TopicTrafficService topicTrafficService;

    @Autowired
    private AuditConsumerConfigService auditConsumerConfigService;

    @Autowired
    private MQProxyService mqProxyService;

    @Autowired
    private AuditHttpConsumerConfigService auditHttpConsumerConfigService;

    /**
     * 消费进度
     * 
     * @param topicParam
     * @return
     * @throws Exception
     */
    @RequestMapping("/progress")
    public String progress(UserInfo userInfo, @RequestParam("tid") int tid,
            @Valid PaginationParam paginationParam, @RequestParam(value = "consumer", required = false) String consumer,
            Map<String, Object> map) throws Exception {
        String view = viewModule() + "/progress";
        // 设置分页参数
        setPagination(map, paginationParam);
        // 获取消费者
        Result<TopicTopology> topicTopologyResult = userService.queryTopicTopology(userInfo.getUser(), tid);
        if (topicTopologyResult.isNotOK()) {
            setResult(map, topicTopologyResult);
            return view;
        }
        // 查询消费进度
        TopicTopology topicTopology = topicTopologyResult.getResult();
        List<Consumer> consumerList = topicTopology.getConsumerList();
        // consumer选择
        ConsumerSelector consumerSelector = new ConsumerSelector();
        if (consumer == null || consumer.length() == 0) {
            pagination(consumerSelector, consumerList, paginationParam);
        } else {
            consumerSelector.select(consumerList, consumer);
        }
        // 获取消费者归属者
        Map<Long, List<User>> consumerMap = getConsumerMap(tid, consumerSelector.getCidList());

        Topic topic = topicTopology.getTopic();
        Cluster cluster = clusterService.getMQClusterById(topic.getClusterId());
        // 获取消费者版本
        Map<String, String> clientVersionMap = queryClientVersionMap(topic.getName(), consumerSelector);
        // 构建集群数据
        List<ConsumerProgressVO> clusterList = buildClusteringConsumerProgressVOList(cluster, topic,
                consumerSelector.getClusteringConsumerList(), consumerMap);
        // 获取消费者配置
        for (ConsumerProgressVO vo : clusterList) {
            vo.setConsumerConfig(consumerConfigService.getConsumerConfig(vo.getConsumer().getName()));
            vo.setVersion(clientVersionMap.get(vo.getConsumer().getName()));
        }
        // 组装集群模式消费者信息
        setResult(map, clusterList);

        // 构建广播数据
        List<ConsumerProgressVO> list = buildBroadcastingConsumerProgressVOList(cluster, topic.getName(),
                consumerSelector.getBroadcastConsumerList(), consumerMap);
        // 获取消费者配置
        for (ConsumerProgressVO vo : list) {
            vo.setConsumerConfig(consumerConfigService.getConsumerConfig(vo.getConsumer().getName()));
            vo.setVersion(clientVersionMap.get(vo.getConsumer().getName()));
            vo.resetClientPauseConfig();
        }
        // 组装广播模式消费者信息
        setResult(map, "resultExt", Result.getResult(list));

        setResult(map, "topic", topic);
        FreemarkerUtil.set("long", Long.class, map);
        setResult(map, "cluster", cluster);

        // 限速常量
        setResult(map, "limitConsumeTps", Constant.LIMIT_CONSUME_TPS);
        // 记录访问足迹
        UserFootprint userFootprint = new UserFootprint();
        userFootprint.setUid(userInfo.getUser().getId());
        userFootprint.setTid(tid);
        userFootprintService.save(userFootprint);
        return view;
    }

    /**
     * 查询消费者版本
     * @param topic
     * @param consumerSelector
     * @return
     */
    private Map<String, String> queryClientVersionMap(String topic, ConsumerSelector consumerSelector) {
        List<String> consumers = consumerSelector.getClusteringConsumerList().stream()
                .map(Consumer::getName)
                .collect(Collectors.toList());
        consumerSelector.getBroadcastConsumerList().stream().map(Consumer::getName).forEach(consumers::add);
        if (consumers.isEmpty()) {
            return Collections.emptyMap();
        }
        Result<List<ClientVersion>> clientVersionListResult = clientVersionService.query(topic, consumers);
        if (clientVersionListResult.isEmpty()) {
            return Collections.emptyMap();
        }
        return clientVersionListResult.getResult().stream()
                .collect(Collectors.toMap(ClientVersion::getClient, ClientVersion::getVersion));
    }

    /**
     * 分页
     * 
     * @param consumerSelector
     * @param consumerList
     * @param paginationParam
     */
    private void pagination(ConsumerSelector consumerSelector, List<Consumer> consumerList,
            PaginationParam paginationParam) {
        paginationParam.caculatePagination(consumerList.size());
        consumerSelector.select(consumerList, paginationParam.getBegin(), paginationParam.getEnd());
    }

    /**
     * 构建集群消费模式消费者消费进度
     * 
     * @param cluster
     * @param topic
     * @param consumerList
     * @param consumerMap
     * @return
     */
    private List<ConsumerProgressVO> buildClusteringConsumerProgressVOList(Cluster cluster, Topic topic,
            List<Consumer> consumerList, Map<Long, List<User>> consumerMap) {
        List<ConsumerProgressVO> list = new ArrayList<ConsumerProgressVO>();
        if (consumerList.isEmpty()) {
            return list;
        }
        Map<Long, ConsumeStats> consumeStatsMap = consumerService.fetchClusteringConsumeProgress(cluster, consumerList);

        // 获取topic上一分钟流量
        Date oneMinuteAgo = new Date(System.currentTimeMillis() - 60000);
        String time = DateUtil.getFormat(DateUtil.HHMM).format(oneMinuteAgo);
        Result<TopicTraffic> trafficResult = topicTrafficService.query(topic.getId(), oneMinuteAgo, time);
        if (trafficResult.isOK()) {
            topic.setCount(trafficResult.getResult().getCount());
        } else {
            topic.setCount(0);
        }

        // 组装集群模式vo
        for (Consumer consumer : consumerList) {
            ConsumerProgressVO consumerProgressVO = buildConsumerProgressVO(topic.getName(), consumerMap, consumer);
            if (consumeStatsMap == null) {
                list.add(consumerProgressVO);
                continue;
            }
            ConsumeStats consumeStats = consumeStatsMap.get(consumer.getId());
            if (consumeStats == null) {
                list.add(consumerProgressVO);
                continue;
            }
            consumerProgressVO.setConsumeTps(consumeStats.getConsumeTps());
            consumerProgressVO.setProduceTps(topic.getCount() / 60);

            // 拆分正常队列和重试队列
            Map<MessageQueue, OffsetWrapper> offsetTable = consumeStats.getOffsetTable();
            Map<MessageQueue, OffsetWrapper> offsetMap = new TreeMap<MessageQueue, OffsetWrapper>();
            Map<MessageQueue, OffsetWrapper> retryOffsetMap = new TreeMap<MessageQueue, OffsetWrapper>();
            for (MessageQueue mq : offsetTable.keySet()) {
                if (CommonUtil.isRetryTopic(mq.getTopic())) {
                    retryOffsetMap.put(mq, offsetTable.get(mq));
                    // 设置topic名字
                    if (consumerProgressVO.getRetryTopic() == null) {
                        consumerProgressVO.setRetryTopic(mq.getTopic());
                    } else if (!mq.getTopic().equals(consumerProgressVO.getRetryTopic())) {
                        logger.error("retry consumer:{} has two diffrent topic, {} != {}", consumer.getName(),
                                mq.getTopic(), consumerProgressVO.getRetryTopic());
                    }
                } else {
                    offsetMap.put(mq, offsetTable.get(mq));
                    if (consumerProgressVO.getTopic() == null) {
                        consumerProgressVO.setTopic(mq.getTopic());
                    } else if (!mq.getTopic().equals(consumerProgressVO.getTopic())) {
                        logger.error("consumer:{} has two diffrent topic, {} != {}", consumer.getName(),
                                mq.getTopic(), consumerProgressVO.getTopic());
                    }
                }
            }
            // 获取死topic状况
            String dlqTopic = MixAll.getDLQTopic(consumer.getName());
            consumerProgressVO.setDlqTopic(dlqTopic);
            TopicStatsTable topicStatsTable = topicService.stats(cluster, dlqTopic);
            if (topicStatsTable != null) {
                consumerProgressVO.setDlqOffsetMap(new TreeMap<>(topicStatsTable.getOffsetTable()));
            }

            consumerProgressVO.setOffsetMap(offsetMap);
            consumerProgressVO.setRetryOffsetMap(retryOffsetMap);
            consumerProgressVO.computeTotalDiff();
            list.add(consumerProgressVO);
        }
        return list;
    }

    /**
     * 构建广播消费模式消费者消费进度
     * 
     * @param cluster
     * @param topic
     * @param consumerList
     * @param consumerMap
     * @return
     */
    private List<ConsumerProgressVO> buildBroadcastingConsumerProgressVOList(Cluster cluster, String topic,
            List<Consumer> consumerList, Map<Long, List<User>> consumerMap) {
        List<ConsumerProgressVO> list = new ArrayList<ConsumerProgressVO>();
        if (consumerList.isEmpty()) {
            return list;
        }
        // 抓取广播消费模式下消费者状态
        Map<Long, List<ConsumeStatsExt>> consumeStatsMap = consumerService.fetchBroadcastingConsumeProgress(cluster,
                topic, consumerList);
        // 组装广播消费模式vo
        for (Consumer consumer : consumerList) {
            ConsumerProgressVO consumerProgressVO = buildConsumerProgressVO(topic, consumerMap, consumer);
            if (consumeStatsMap == null) {
                list.add(consumerProgressVO);
                continue;
            }
            List<ConsumeStatsExt> consumeStatsList = consumeStatsMap.get(consumer.getId());
            if (consumeStatsList == null) {
                list.add(consumerProgressVO);
                continue;
            }
            Map<MessageQueue, OffsetWrapper> offsetMap = new TreeMap<MessageQueue, OffsetWrapper>();
            for (ConsumeStatsExt consumeStats : consumeStatsList) {
                consumerProgressVO.setConsumeTps(consumerProgressVO.getConsumeTps() + consumeStats.getConsumeTps());
                Map<MessageQueue, OffsetWrapper> offsetTable = consumeStats.getOffsetTable();
                for (MessageQueue mq : offsetTable.keySet()) {
                    OffsetWrapper prev = offsetMap.get(mq);
                    if (prev == null) {
                        prev = new OffsetWrapper();
                        offsetMap.put(mq, prev);
                    }
                    OffsetWrapper cur = offsetTable.get(mq);
                    if (cur.getConsumerOffset() < 0) {
                        cur.setConsumerOffset(0);
                    }
                    prev.setBrokerOffset(prev.getBrokerOffset() + cur.getBrokerOffset());
                    prev.setConsumerOffset(prev.getConsumerOffset() + cur.getConsumerOffset());
                    // 取最小的更新时间
                    if (prev.getLastTimestamp() == 0 || prev.getLastTimestamp() > cur.getLastTimestamp()) {
                        prev.setLastTimestamp(cur.getLastTimestamp());
                    }
                }
            }
            consumerProgressVO.setOffsetMap(offsetMap);
            consumerProgressVO.setConsumeStatsList(consumeStatsList);
            consumerProgressVO.computeTotalDiff();
            list.add(consumerProgressVO);
        }
        return list;
    }

    /**
     * 构建ConsumerProgressVO
     * 
     * @return
     */
    public ConsumerProgressVO buildConsumerProgressVO(String topic, Map<Long, List<User>> consumerMap,
            Consumer consumer) {
        ConsumerProgressVO consumerProgressVO = new ConsumerProgressVO();
        consumerProgressVO.setConsumer(consumer);
        consumerProgressVO.setTopic(topic);
        consumerProgressVO.setOwnerList(consumerMap.get(consumer.getId()));
        return consumerProgressVO;
    }

    /**
     * 获取消费者map
     * 
     * @param tid
     * @param cidList
     * @return
     */
    private Map<Long, List<User>> getConsumerMap(long tid, List<Long> cidList) {
        Map<Long, List<User>> consumerMap = new HashMap<Long, List<User>>();
        if (cidList == null || cidList.size() == 0) {
            return consumerMap;
        }
        Result<List<UserConsumer>> ucListResult = userConsumerService.queryTopicConsumer(tid, cidList);
        if (ucListResult.isEmpty()) {
            return consumerMap;
        }
        Set<Long> uidList = new HashSet<Long>();
        for (UserConsumer userConsumer : ucListResult.getResult()) {
            uidList.add(userConsumer.getUid());
        }
        Result<List<User>> userListResult = userService.query(uidList);
        if (userListResult.isEmpty()) {
            return consumerMap;
        }
        for (UserConsumer userConsumer : ucListResult.getResult()) {
            for (User user : userListResult.getResult()) {
                if (userConsumer.getUid() == user.getId()) {
                    List<User> userList = consumerMap.get(userConsumer.getConsumerId());
                    if (userList == null) {
                        userList = new ArrayList<User>();
                        consumerMap.put(userConsumer.getConsumerId(), userList);
                    }
                    userList.add(user);
                }
            }
        }
        return consumerMap;
    }

    /**
     * 重置偏移量
     * 
     * @param userConsumerParam
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping("/resetOffset")
    public Result<?> resetOffset(UserInfo userInfo, @Valid UserConsumerParam userConsumerParam) throws Exception {
        // 校验用户是否能重置，防止误调用接口
        Result<List<UserConsumer>> userConsumerListResult = userConsumerService.queryUserConsumer(userInfo.getUser(),
                userConsumerParam.getTid(), userConsumerParam.getConsumerId());
        if (userConsumerListResult.isEmpty() && !userInfo.getUser().isAdmin()) {
            return Result.getResult(Status.PERMISSION_DENIED_ERROR);
        }
        Date resetOffsetTo = null;
        try {
            if (userConsumerParam.getOffset() != null) {
                resetOffsetTo = DateUtil.getFormat(DateUtil.YMD_DASH_BLANK_HMS_COLON)
                        .parse(userConsumerParam.getOffset());
            }
        } catch (Exception e) {
            logger.error("resetOffsetTo param err:{}", userConsumerParam.getOffset(), e);
            return Result.getResult(Status.PARAM_ERROR);
        }

        // 构造审核记录
        Audit audit = new Audit();
        // 重新定义操作成功返回的文案
        String message = "";
        if (resetOffsetTo == null) {
            audit.setType(TypeEnum.RESET_OFFSET_TO_MAX.getType());
            message = "跳过堆积申请成功！请耐心等待！";
        } else {
            audit.setType(TypeEnum.RESET_OFFSET.getType());
            message = "消息回溯申请成功！请耐心等待！";
        }
        audit.setUid(userInfo.getUser().getId());
        // 构造重置对象
        AuditResetOffset auditResetOffset = new AuditResetOffset();
        BeanUtils.copyProperties(userConsumerParam, auditResetOffset);
        // 保存记录
        Result<?> result = auditService.saveAuditAndSkipAccumulation(audit, auditResetOffset);
        if (result.isOK()) {
            String tip = getTopicConsumerTip(userConsumerParam.getTid(), userConsumerParam.getConsumerId());
            alertService.sendAuditMail(userInfo.getUser(), TypeEnum.getEnumByType(audit.getType()), tip);
            // 重新定义返回文案
            result.setMessage(message);
        }
        return Result.getWebResult(result);
    }

    /**
     * 删除消费者
     * 
     * @param topicParam
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping(value = "/delete", method = RequestMethod.POST)
    public Result<?> delete(UserInfo userInfo, @Valid UserConsumerParam userConsumerParam) throws Exception {
        Result<?> isExist = verifyDataService.verifyDeleteConsumerIsExist(userConsumerParam);
        if (isExist.getStatus() != Status.OK.getKey()) {
            return isExist;
        }
        // 校验用户是否能删除，防止调用接口误删
        Result<List<UserConsumer>> userConsumerListResult = userConsumerService.queryUserConsumer(userInfo.getUser(),
                userConsumerParam.getTid(), userConsumerParam.getConsumerId());
        if (userConsumerListResult.isEmpty() && !userInfo.getUser().isAdmin()) {
            return Result.getResult(Status.PERMISSION_DENIED_ERROR);
        }

        Result<Consumer> consumerResult = consumerService.queryById(userConsumerParam.getConsumerId());
        if (consumerResult.isNotOK()) {
            return Result.getWebResult(consumerResult);
        }

        Result<Topic> topicResult = topicService.queryTopic(consumerResult.getResult().getTid());
        if (topicResult.isNotOK()) {
            return Result.getWebResult(topicResult);
        }

        // 校验是否还有链接
        String consumer = consumerResult.getResult().getName();
        if (!consumerResult.getResult().isHttpProtocol()) {
            Cluster cluster = clusterService.getMQClusterById(topicResult.getResult().getClusterId());
            Result<ConsumerConnection> connectionResult = consumerService.examineConsumerConnectionInfo(consumer,
                    cluster, consumerResult.getResult().isProxyRemoting());
            if (connectionResult.isOK()) {
                return Result.getResult(Status.CONSUMER_CONNECTION_EXIST_ERROR);
            }
        }

        // 构造审核记录
        Audit audit = new Audit();
        audit.setType(TypeEnum.DELETE_CONSUMER.getType());
        audit.setUid(userInfo.getUser().getId());
        // 保存记录
        Result<?> result = auditService.saveAuditAndConsumerDelete(audit, userConsumerParam.getConsumerId(),
                consumer, topicResult.getResult().getName());
        if (result.isOK()) {
            alertService.sendAuditMail(userInfo.getUser(), TypeEnum.DELETE_CONSUMER,
                    consumerResult.getResult().getName());
        }
        return Result.getWebResult(result);
    }

    /**
     * 消费者列表
     * 
     * @param topicParam
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping("/list")
    public Result<?> list(UserInfo userInfo, @RequestParam("tid") int tid) throws Exception {
        Result<List<Consumer>> consumerListResult = consumerService.queryByTid(tid);
        return Result.getWebResult(consumerListResult);
    }

    /**
     * 消费者列表
     * 
     * @param topicParam
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping("/list/all")
    public Result<?> listAll(UserInfo userInfo) throws Exception {
        Result<List<Consumer>> consumerListResult = consumerService.queryAll();
        return Result.getWebResult(consumerListResult);
    }

    @RequestMapping(value = "/associate", method = RequestMethod.GET)
    public String associate(Map<String, Object> map)
            throws Exception {
        setView(map, "associate", "关联消费者");
        return view();
    }

    /**
     * 关联消费者
     * 
     * @param topicParam
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping(value = "/associate", method = RequestMethod.POST)
    public Result<?> associate(UserInfo userInfo, @Valid AssociateConsumerParam associateConsumerParam)
            throws Exception {
        return associateUserConsumer(userInfo, userInfo.getUser().getId(), associateConsumerParam.getTid(),
                associateConsumerParam.getCid());
    }

    /**
     * 授权关联
     * 
     * @param userInfo
     * @param tid
     * @param uid
     * @param cid
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping(value = "/auth/associate", method = RequestMethod.POST)
    public Result<?> authAssociate(UserInfo userInfo, @RequestParam("tid") long tid,
            @RequestParam("uid") long uid,
            @RequestParam("cid") long cid) throws Exception {
        if (tid < 1 || uid < 1 || cid < 1) {
            return Result.getResult(Status.PARAM_ERROR);
        }
        return associateUserConsumer(userInfo, uid, tid, cid);
    }

    /**
     * 复用之前的逻辑
     * 
     * @param userInfo
     * @param uid
     * @param tid
     * @param cid
     * @return
     */
    private Result<?> associateUserConsumer(UserInfo userInfo, long uid, long tid, long cid) {
        // 验证用户是否已经关联过此消费者
        Result<?> isExist = verifyDataService.verifyUserConsumerIsExist(uid, cid);
        if (isExist.getStatus() != Status.OK.getKey()) {
            return isExist;
        }
        AuditAssociateConsumer auditAssociateConsumer = new AuditAssociateConsumer();
        auditAssociateConsumer.setCid(cid);
        auditAssociateConsumer.setTid(tid);
        auditAssociateConsumer.setUid(uid);
        // 构建Audit
        Audit audit = new Audit();
        audit.setType(Audit.TypeEnum.ASSOCIATE_CONSUMER.getType());
        audit.setStatus(Audit.StatusEnum.INIT.getStatus());
        audit.setUid(userInfo.getUser().getId());
        Result<Audit> result = auditService.saveAuditAndAssociateConsumer(audit, auditAssociateConsumer);
        if (result.isOK()) {
            String tip = getTopicConsumerTip(tid, cid);
            if (uid != userInfo.getUser().getId()) {
                Result<User> userResult = userService.query(uid);
                if (userResult.isNotOK()) {
                    return Result.getResult(Status.EMAIL_SEND_ERR);
                }
                tip = tip + " user:<b>" + userResult.getResult().notBlankName() + "</b>";
            }
            alertService.sendAuditMail(userInfo.getUser(), TypeEnum.ASSOCIATE_CONSUMER, tip);
        }
        return Result.getWebResult(result);
    }

    @RequestMapping(value = "/add", method = RequestMethod.GET)
    public String add(Map<String, Object> map)
            throws Exception {
        setView(map, "add", "新建消费者");
        return view();
    }

    /**
     * 新建消费者
     * 
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping(value = "/add", method = RequestMethod.POST)
    public Result<?> add(UserInfo userInfo, @Valid ConsumerParam consumerParam) throws Exception {
        logger.info("create consumer, user:{} consumerParam:{}", userInfo, consumerParam);
        Result<?> isExist = verifyDataService.verifyAddConsumerIsExist(userInfo.getUser().getId(),
                consumerParam.getConsumer());
        if (isExist.isNotOK()) {
            return isExist;
        }
        // 构建审核记录
        Audit audit = new Audit();
        audit.setType(Audit.TypeEnum.NEW_CONSUMER.getType());
        audit.setStatus(Audit.StatusEnum.INIT.getStatus());
        audit.setUid(userInfo.getUser().getId());
        audit.setInfo(consumerParam.getInfo());

        // 构建消费者审核记录
        AuditConsumer auditConsumer = new AuditConsumer();
        BeanUtils.copyProperties(consumerParam, auditConsumer);
        // 保存记录
        Result<?> result = auditService.saveAuditAndConsumer(audit, auditConsumer);
        if (result.isOK()) {
            String topicTip = getTopicTip(consumerParam.getTid());
            alertService.sendAuditMail(userInfo.getUser(), TypeEnum.NEW_CONSUMER,
                    topicTip + " consumer:<b>" + consumerParam.getConsumer() + "</b>");
        }
        return Result.getWebResult(result);
    }

    /**
     * 根据broker查询队列信息
     * 
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping(value = "/broker/queue", method = RequestMethod.GET)
    public Result<?> getQueueList(@RequestParam("topic") String topic,
            @RequestParam("brokerName") String brokerName,
            @RequestParam("clusterId") long clusterId) throws Exception {
        if (topic == "" || brokerName == "") {
            return Result.getResult(Status.PARAM_ERROR);
        }
        Cluster cluster = clusterService.getMQClusterById(clusterId);
        if (cluster == null) {
            return Result.getResult(cluster);
        }
        TopicStatsTable topicStatsTable = topicService.stats(cluster, topic);
        if (topicStatsTable == null) {
            return Result.getResult(topicStatsTable);
        }
        Map<Integer, Long> queueOffsetMap = new TreeMap<Integer, Long>();
        for (MessageQueue mq : topicStatsTable.getOffsetTable().keySet()) {
            if (mq.getBrokerName().equals(brokerName)) {
                long maxOffset = topicStatsTable.getOffsetTable().get(mq).getMaxOffset();
                if (!queueOffsetMap.containsKey(mq.getQueueId()) || queueOffsetMap.get(mq.getQueueId()) < maxOffset) {
                    queueOffsetMap.put(mq.getQueueId(), maxOffset);
                }
            }
        }
        return Result.getResult(queueOffsetMap);
    }

    /**
     * 获取topic和consumer的提示信息
     * 
     * @param tid
     * @param cid
     * @return
     */
    private String getTopicConsumerTip(long tid, long cid) {
        Result<Topic> topicResult = topicService.queryTopic(tid);
        String topic = null;
        if (topicResult.isOK()) {
            topic = topicResult.getResult().getName();
        }
        Result<Consumer> consumerResult = consumerService.queryById(cid);
        String consumer = null;
        if (consumerResult.isOK()) {
            consumer = consumerResult.getResult().getName();
        }
        return getTopicConsumerTip(topic, consumer);
    }

    private String getTopicConsumerTip(String topic, String consumer) {
        StringBuilder sb = new StringBuilder();
        if (topic != null) {
            sb.append(" topic:<b>");
            sb.append(topic);
            sb.append("</b>");
        }
        if (consumer != null) {
            sb.append(" consumer:");
            if (topic != null) {
                sb.append(mqCloudConfigHelper.getTopicConsumeHrefLink(topic, consumer));
            } else {
                sb.append(consumer);
            }
        }
        return sb.toString();
    }

    /**
     * 获取topic的提示信息
     * 
     * @param tid
     * @return
     */
    private String getTopicTip(long tid) {
        StringBuilder sb = new StringBuilder();
        Result<Topic> topicResult = topicService.queryTopic(tid);
        if (topicResult.isOK()) {
            sb.append(" topic:<b>");
            sb.append(topicResult.getResult().getName());
            sb.append("</b>");
        }
        return sb.toString();
    }

    /**
     * 获取queue的拥有者
     * 
     * @param userInfo
     * @param cid
     * @param consumer
     * @return
     * @throws Exception
     */
    @ResponseBody
    @RequestMapping("/queue/owner")
    public Result<?> queueOwner(UserInfo userInfo, @RequestParam("cid") int cid,
                                @RequestParam("consumer") String consumerName) throws Exception {
        Cluster cluster = clusterService.getMQClusterById(cid);
        // 获取消费者运行时信息
        Consumer consumer = consumerService.queryConsumerByName(consumerName).getResult();
        Map<String, ConsumerRunningInfo> map = consumerService.getConsumerRunningInfo(cluster, consumerName,
                consumer.isProxyRemoting());
        if (map == null) {
            return Result.getResult(Status.NO_RESULT);
        }
        ConsumerConfig consumerConfig = consumerConfigService.getConsumerConfig(consumerName);
        // 组装vo
        List<QueueOwnerVO> queueConsumerVOList = new ArrayList<>();
        for (String clientId : map.keySet()) {
            boolean addConsumerRunningInfo = false;
            String ip = clientId.substring(0, clientId.indexOf("@"));
            ConsumerRunningInfo consumerRunningInfo = map.get(clientId);
            for (MessageQueue messageQueue : consumerRunningInfo.getMqTable().keySet()) {
                QueueOwnerVO queueOwnerVO = new QueueOwnerVO();
                if (!addConsumerRunningInfo) {
                    addConsumerRunningInfo = setConsumerRunningInfo(messageQueue, consumerRunningInfo, queueOwnerVO);
                }
                BeanUtils.copyProperties(messageQueue, queueOwnerVO);
                queueOwnerVO.setClientId(clientId);
                queueOwnerVO.setIp(ip);
                setPauseInfo(consumerConfig, queueOwnerVO);
                queueConsumerVOList.add(queueOwnerVO);
            }
        }
        return Result.getResult(queueConsumerVOList);
    }

    /**
     * 设置消费者运行时信息
     * @param messageQueue
     * @param consumerRunningInfo
     * @param queueOwnerVO
     * @return
     */
    private boolean setConsumerRunningInfo(MessageQueue messageQueue, ConsumerRunningInfo consumerRunningInfo, QueueOwnerVO queueOwnerVO){
        if (CommonUtil.isRetryTopic(messageQueue.getTopic()) || CommonUtil.isDeadTopic(messageQueue.getTopic())) {
            return false;
        }
        ConsumerRunningInfoVO consumerRunningInfoVO = new ConsumerRunningInfoVO(consumerRunningInfo);
        queueOwnerVO.setShowClientInfo(true);
        queueOwnerVO.setWarnInfo(consumerRunningInfoVO.getWarnInfo());
        queueOwnerVO.setLanguage(consumerRunningInfoVO.getLanguage());
