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

import com.sohu.tv.mq.cloud.bo.*;
import com.sohu.tv.mq.cloud.bo.Audit.StatusEnum;
import com.sohu.tv.mq.cloud.bo.Audit.TypeEnum;
import com.sohu.tv.mq.cloud.service.*;
import com.sohu.tv.mq.cloud.service.MQProxyService.ConsumerConfigParam;
import com.sohu.tv.mq.cloud.util.DateUtil;
import com.sohu.tv.mq.cloud.util.MQCloudConfigHelper;
import com.sohu.tv.mq.cloud.util.Result;
import com.sohu.tv.mq.cloud.util.Status;
import com.sohu.tv.mq.cloud.web.controller.param.PaginationParam;
import com.sohu.tv.mq.cloud.web.vo.*;
import com.sohu.tv.mq.util.CommonUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.protocol.body.Connection;
import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.validation.Valid;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.RejectedExecutionException;

import static com.sohu.tv.mq.cloud.bo.Audit.StatusEnum.AUDITING;
import static com.sohu.tv.mq.cloud.bo.Audit.StatusEnum.INIT;
import static com.sohu.tv.mq.cloud.util.Status.WEB_ERROR;

/**
 * 审核
 * 
 * @Description:
 * @author yongfeigao
 * @date 2018年7月5日
 */
@Controller
@RequestMapping("/admin/audit")
public class AdminAuditController extends AdminViewController {

    @Autowired
    private AuditService auditService;

    @Autowired
    private AuditTopicService auditTopicService;

    @Autowired
    private UserService userService;

    @Autowired
    private TopicService topicService;

    @Autowired
    private UserMessageService userMessageService;

    @Autowired
    private AssociateProducerService associateProducerService;

    @Autowired
    private AssociateConsumerService associateConsumerService;

    @Autowired
    private UserProducerService userProducerService;

    @Autowired
    private UserConsumerService userConsumerService;

    @Autowired
    private ConsumerService consumerService;

    @Autowired
    private AuditConsumerService auditConsumerService;

    @Autowired
    private AuditTopicDeleteService auditTopicDeleteService;

    @Autowired
    private AuditConsumerDeleteService auditConsumerDeleteService;

    @Autowired
    private AuditResetOffsetService auditResetOffsetService;

    @Autowired
    private AuditTopicUpdateService auditTopicUpdateService;

    @Autowired
    private AuditUserProducerDeleteService auditUserProducerDeleteService;

    @Autowired
    private AuditUserConsumerDeleteService auditUserConsumerDeleteService;

    @Autowired
    private ClusterService clusterService;

    @Autowired
    private AuditResendMessageService auditResendMessageService;

    @Autowired
    private AlertService alertService;

    @Autowired
    private ConsumerConfigService consumerConfigService;

    @Autowired
    private AuditTopicTraceService auditTopicTraceService;

    @Autowired
    private AuditConsumerConfigService auditConsumerConfigService;

    @Autowired
    private AuditTopicTrafficWarnService auditTopicTrafficWarnService;

    @Autowired
    private AuditTimespanMessageConsumeService auditTimespanMessageConsumeService;

    @Autowired
    private MQProxyService mqProxyService;

    @Autowired
    private AuditWheelMessageCancelService auditWheelMessageCancelService;

    @Autowired
    private MQCloudConfigHelper mqCloudConfigHelper;

    @Autowired
    private AuditTimespanMessageExportService auditTimespanMessageExportService;

    @Autowired
    private MessageExportService messageExportService;

    @Autowired
    private AuditHttpConsumerConfigService auditHttpConsumerConfigService;

    /**
     * 审核主列表
     * 
     * @param type
     * @param status
     * @param map
     * @return
     */
    @RequestMapping("/list")
    public String list(@RequestParam(value = "type", defaultValue = "-1") int type,
            @RequestParam(value = "status", defaultValue = "0") int status, @Valid PaginationParam paginationParam,
            Map<String, Object> map) {
        // 设置返回视图及常量
        setView(map, "list");
        setResult(map, "status", Audit.StatusEnum.values());
        setResult(map, "type", Audit.TypeEnum.values());
        Result.setResult(map, paginationParam);

        // 构造查询参数
        Audit audit = new Audit();
        audit.setType(type);
        audit.setStatus(status);

        // 获取审核列表数量
        Result<Integer> countResult = auditService.queryCount(audit);
        if (!countResult.isOK()) {
            return view();
        }
        paginationParam.caculatePagination(countResult.getResult());
        // 查询审核列表
        Result<List<Audit>> auditListResult = auditService.queryAuditListByPage(audit, paginationParam.getBegin(),
                paginationParam.getNumOfPage());
        if (auditListResult.isEmpty()) {
            return view();
        }

        // 拼装VO
        List<Audit> auditList = auditListResult.getResult();
        List<AuditVO> auditVOList = new ArrayList<AuditVO>(auditList.size());
        Set<Long> idSet = new HashSet<Long>(auditList.size());
        for (Audit auditObj : auditList) {
            AuditVO auditVo = new AuditVO();
            BeanUtils.copyProperties(auditObj, auditVo);
            auditVo.setStatusEnum(StatusEnum.getEnumByStatus(auditObj.getStatus()));
            auditVo.setTypeEnum(TypeEnum.getEnumByType(auditObj.getType()));
            auditVOList.add(auditVo);

            idSet.add(auditObj.getUid());
        }

        // 查询用户
        Result<List<User>> userListResult = userService.query(idSet);
        if (userListResult.isNotEmpty()) {
            // vo赋值
            assgin(userListResult.getResult(), auditVOList);
        }
        setResult(map, auditVOList);
        return view();
    }

    /**
     * 申请详情
     * 
     * @param aid
     * @param map
     * @return
     */
    @RequestMapping("/detail")
    public String detail(@RequestParam("type") int type,
            @RequestParam(value = "aid") long aid, Map<String, Object> map) {
        TypeEnum typeEnum = TypeEnum.getEnumByType(type);
        // 创建topic需要集群信息
        if (TypeEnum.NEW_TOPIC == typeEnum || TypeEnum.UPDATE_TOPIC_TRACE == typeEnum) {
            setResult(map, "clusters", clusterService.getAllMQCluster());
        }
        Result<?> result = auditService.detail(typeEnum, aid);
        setResult(map, result);
        return adminViewModule() + "/" + typeEnum.getView();
    }

    /**
     * 拒绝
     * 
     * @param aid
     * @param map
     * @return
     */
    @ResponseBody
    @RequestMapping(value = "/refuse", method = RequestMethod.POST)
    public Result<?> refuse(UserInfo userInfo,
            @RequestParam("aid") long aid,
            @RequestParam(value = "refuseReason") String refuseReason) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        // 校验状态是否合法
        Audit audit = auditResult.getResult();
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }

        // 拒绝
        Audit updateAudit = new Audit();
        updateAudit.setId(audit.getId());
        updateAudit.setAuditor(userInfo.getUser().getEmail());
        updateAudit.setStatus(StatusEnum.REJECT.getStatus());
        updateAudit.setRefuseReason(refuseReason);
        Result<Integer> updateResult = auditService.updateAudit(updateAudit, INIT.getStatus());
        if (updateResult.isOK()) {
            // 拼装提示消息
            TypeEnum typeEnum = TypeEnum.getEnumByType(audit.getType());
            String msg = null;
            switch (typeEnum) {
                case NEW_TOPIC:
                    // 获取AuditTopic
                    Result<AuditTopic> auditTopicResult = auditTopicService.queryAuditTopic(aid);
                    if (auditTopicResult.isOK()) {
                        msg = auditTopicResult.getResult().getName();
                    }
                    break;
                case UPDATE_TOPIC:
                    msg = getUpdateTopicTipMessage(aid);
                    break;
                case DELETE_TOPIC:
                    msg = getDeleteTopicTipMessage(aid);
                    break;
                case NEW_CONSUMER:
                    Result<AuditConsumer> auditConsumerResult = auditConsumerService.queryAuditConsumer(aid);
                    if (auditConsumerResult.isOK()) {
                        msg = auditConsumerResult.getResult().getConsumer();
                    }
                    break;
                case DELETE_CONSUMER:
                    msg = getDeleteConsumerTipMessage(aid);
                    break;
                case RESET_OFFSET:
                case RESET_OFFSET_TO_MAX:
                case RESET_RETRY_OFFSET:
                    msg = getAuditResetOffsetTipMessage(aid);
                    break;
                case ASSOCIATE_PRODUCER:
                case NEW_PRODUCER:
                    msg = getAuditAssociateProducerTipMessage(aid);
                    break;
                case ASSOCIATE_CONSUMER:
                    msg = getAuditAssociateConsumerTipMessage(aid);
                    break;
                case BECOME_ADMIN:
                    break;
                case DELETE_USERPRODUCER:
                    msg = getDeleteUserProducerTipMessage(aid);
                    break;
                case DELETE_USERCONSUMER:
                    msg = getDeleteUserConsumerTipMessage(aid);
                    break;
                case RESEND_MESSAGE:
                    msg = null;
                    break;
                case UPDATE_TOPIC_TRACE:
                    msg = getUpdateTopicTraceMessage(aid);
                    break;
                case BATCH_ASSOCIATE:
                    break;
                case PAUSE_CONSUME:
                case RESUME_CONSUME:
                case LIMIT_CONSUME:
                case UPDATE_HTTP_CONSUMER_CONFIG:
                    msg = getConsumerConfigMessage(aid);
                    break;
                case UPDATE_TOPIC_TRAFFIC_WARN:
                    msg = getUpdateTopicTrafficWarnMessage(aid);
                    break;
                case TIMESPAN_MESSAGE_CONSUME:
                    msg = getTimespanMessageConsumeResult(aid);
                    break;
            }
            StringBuilder sb = new StringBuilder("您");
            if (audit.getCreateTime() != null) {
                String createTime = DateUtil.getFormat(DateUtil.YMD_DASH_BLANK_HMS_COLON).format(audit.getCreateTime());
                sb.append(createTime);
            }
            sb.append("申请的");
            sb.append(typeEnum.getName());
            if (msg != null) {
                sb.append("[");
                sb.append(msg);
                sb.append("]");
            }
            sb.append("被拒绝。原因:");
            sb.append(refuseReason);
            UserMessage userMessage = new UserMessage();
            userMessage.setMessage(sb.toString());
            userMessage.setUid(audit.getUid());
            userMessageService.save(userMessage);
            sendEmailMessage(audit.getUid(), sb.toString());
            return updateResult;
        }
        return Result.getResult(Status.PARAM_ERROR);
    }

    /**
     * 获取时间段消费
     *
     * @param aid
     * @return Result
     */
    private String getTimespanMessageConsumeResult(long aid) {
        Result<AuditTimespanMessageConsume> result = auditTimespanMessageConsumeService.query(aid);
        if (result.isNotOK()) {
            return null;
        }
        return result.getResult().getTopic() + "(" + result.getResult().getStartFormat() + ","
                + result.getResult().getEndFormat() + ")";
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    private String getAuditAssociateProducerTipMessage(long aid) {
        Result<AuditAssociateProducer> auditAssociateProducerResult = associateProducerService.query(aid);
        if (auditAssociateProducerResult.isOK()) {
            return auditAssociateProducerResult.getResult().getProducer();
        }
        return null;
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    private String getAuditResetOffsetTipMessage(long aid) {
        Result<AuditResetOffset> auditResetOffsetResult = auditResetOffsetService.queryAuditResetOffset(aid);
        if (auditResetOffsetResult.isNotOK()) {
            return null;
        }
        // 查询consumer信息
        Result<Consumer> consumerResult = consumerService.queryById(auditResetOffsetResult.getResult().getConsumerId());
        if (consumerResult.isNotOK()) {
            return null;
        }
        return consumerResult.getResult().getName();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    public String getUpdateTopicTipMessage(long aid) {
        Result<AuditTopicUpdate> auditTopicUpdateResult = auditTopicUpdateService.queryAuditTopicUpdate(aid);
        if (auditTopicUpdateResult.isNotOK()) {
            return null;
        }
        AuditTopicUpdate auditTopicUpdate = auditTopicUpdateResult.getResult();
        // 查询topic记录
        Result<Topic> topicResult = topicService.queryTopic(auditTopicUpdate.getTid());
        if (topicResult.isNotOK()) {
            return null;
        }
        return topicResult.getResult().getName();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    public String getUpdateTopicTraceMessage(long aid) {
        Result<AuditTopicTrace> auditTopicTraceResult = auditTopicTraceService.queryAuditTopicTrace(aid);
        if (auditTopicTraceResult.isNotOK()) {
            return null;
        }
        AuditTopicTrace auditTopicTrace = auditTopicTraceResult.getResult();
        // 查询topic记录
        Result<Topic> topicResult = topicService.queryTopic(auditTopicTrace.getTid());
        if (topicResult.isNotOK()) {
            return null;
        }
        return topicResult.getResult().getName();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    public String getDeleteTopicTipMessage(long aid) {
        // 查询 topic删除审核记录
        Result<AuditTopicDelete> auditTopicDeleteResult = auditTopicDeleteService.queryAuditTopicDelete(aid);
        if (auditTopicDeleteResult.isNotOK()) {
            return null;
        }
        AuditTopicDelete auditTopicDelete = auditTopicDeleteResult.getResult();
        // 查询topic记录
        Result<Topic> topicResult = topicService.queryTopic(auditTopicDelete.getTid());
        if (topicResult.isNotOK()) {
            return null;
        }
        return topicResult.getResult().getName();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    public String getDeleteConsumerTipMessage(long aid) {
        // 查询consumer删除审核记录
        Result<AuditConsumerDelete> auditConsumerDeleteResult = auditConsumerDeleteService
                .queryAuditConsumerDelete(aid);
        if (auditConsumerDeleteResult.isNotOK()) {
            return null;
        }
        AuditConsumerDelete auditConsumerDelete = auditConsumerDeleteResult.getResult();
        // 查询consumer信息
        Result<Consumer> consumerResult = consumerService.queryById(auditConsumerDelete.getCid());
        if (consumerResult.isNotOK()) {
            return null;
        }
        return consumerResult.getResult().getName();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    private String getAuditAssociateConsumerTipMessage(long aid) {
        Result<AuditAssociateConsumer> auditAssociateConsumerResult = associateConsumerService.query(aid);
        if (auditAssociateConsumerResult.isNotOK()) {
            return null;
        }
        Result<Consumer> consumerResult = consumerService.queryById(auditAssociateConsumerResult.getResult().getCid());
        if (consumerResult.isNotOK()) {
            return null;
        }
        return consumerResult.getResult().getName();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    public String getDeleteUserProducerTipMessage(long aid) {
        // 查询consumer删除审核记录
        Result<AuditUserProducerDelete> auditUserProducerDeleteResult = auditUserProducerDeleteService
                .queryAuditUserProducerDelete(aid);
        if (auditUserProducerDeleteResult.isNotOK()) {
            return null;
        }
        AuditUserProducerDelete auditUserProducerDelete = auditUserProducerDeleteResult.getResult();
        // 查询UserProducer信息
        Result<UserProducer> userProducerResult = userProducerService
                .findUserProducer(auditUserProducerDelete.getPid());
        if (userProducerResult.isNotOK()) {
            return null;
        }
        return userProducerResult.getResult().getProducer();
    }

    /**
     * 获取提示消息
     * 
     * @param aid
     * @return
     */
    public String getDeleteUserConsumerTipMessage(long aid) {
        // 查询consumer删除审核记录
        Result<AuditUserConsumerDelete> auditUserConsumerDeleteResult = auditUserConsumerDeleteService
                .queryAuditUserConsumerDelete(aid);
        if (auditUserConsumerDeleteResult.isNotOK()) {
            return null;
        }
        AuditUserConsumerDelete auditUserConsumerDelete = auditUserConsumerDeleteResult.getResult();
        // 查询Consumer信息
        Result<Consumer> consumerResult = consumerService.queryConsumerByName(auditUserConsumerDelete.getConsumer());
        if (consumerResult.isNotOK()) {
            return null;
        }
        return consumerResult.getResult().getName();
    }

    /**
     * 成为管理员
     * 
     * @param aid
     * @param map
     * @return
     */
    @ResponseBody
    @RequestMapping(value = "/becomeAdmin", method = RequestMethod.POST)
    public Result<?> becomeAdmin(UserInfo userInfo,
            @RequestParam("aid") long aid) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        // 校验是否合法
        Audit audit = auditResult.getResult();
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }

        // 设置管理员
        User user = new User();
        user.setId(audit.getUid());
        user.setType(User.ADMIN);
        // 更新用户
        Result<?> result = userService.update(user);
        if (result.isNotOK()) {
            return Result.getWebResult(result);
        }

        // 更新申请状态
        boolean updateOK = agreeAndTip(audit, userInfo.getUser().getEmail(), null);
        if (updateOK) {
            return Result.getOKResult();
        }
        return Result.getResult(Status.DB_UPDATE_ERR_USER_UPDATE_OK);
    }

    /**
     * 创建topic
     * 
     * @param userInfo
     * @param aid
     * @param cid
     * @param traceClusterId为要创建trace topic的集群id，可为空
     * @return
     */
    @ResponseBody
    @RequestMapping(value = "/topic/create", method = RequestMethod.POST)
    public Result<?> createTopic(UserInfo userInfo,
            @RequestParam("aid") long aid, @RequestParam("cid") int cid,
            @RequestParam(value = "traceClusterId", defaultValue = "0") Integer traceClusterId) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        // 校验是否合法
        Audit audit = auditResult.getResult();
        if (TypeEnum.NEW_TOPIC.getType() != audit.getType()) {
            return Result.getResult(Status.PARAM_ERROR);
        }
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }

        // 获取AuditTopic
        Result<AuditTopic> auditTopicResult = auditTopicService.queryAuditTopic(aid);
        if (auditTopicResult.isNotOK()) {
            return auditTopicResult;
        }
        // 获取集群
        Result<Cluster> mqClusterResult = clusterService.queryById(cid);
        if (mqClusterResult.isNotOK()) {
            return Result.getResult(Status.PARAM_ERROR);
        }
        Cluster cluster = mqClusterResult.getResult();
        if (cluster.isBrokerUpdating()) {
            return Result.getResult(Status.BROKER_UPDATING);
        }

        AuditTopic auditTopic = auditTopicResult.getResult();
        // 创建topic
        Result<?> createResult = topicService.createTopic(cluster, audit, auditTopic);
        if (createResult.isNotOK()) {
            return createResult;
        }
        // 更新申请状态
        boolean updateOK = agreeAndTip(audit, userInfo.getUser().getEmail(), auditTopic.getName());
        if (updateOK) {
            // 将trace topic的创建置后，不影响主流程，出错后需要手动创建
            if (auditTopic.traceEnabled()) {
                Result<?> traceTopicResult = topicService.createTraceTopic(audit, auditTopic, traceClusterId);
                if (traceTopicResult.isNotOK()) {
                    return Result.getResult(Status.TOPIC_CREATE_OK_BUT_TRACE_TOPIC_CREATE_ERROR);
                }
            }
            return Result.getOKResult();
        }
        return Result.getResult(Status.DB_UPDATE_ERR_TOPIC_OK);
    }

    /**
     * 创建消费者
     */
    @ResponseBody
    @RequestMapping(value = "/addConsumer", method = RequestMethod.POST)
    public Result<?> addConsumer(UserInfo userInfo,
            @RequestParam("aid") long aid) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        Audit audit = auditResult.getResult();
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }
        // 获取AuditConsumer
        Result<AuditConsumer> auditConsumerResult = auditConsumerService.queryAuditConsumer(aid);
        if (auditConsumerResult.isNotOK()) {
            return auditConsumerResult;
        }
        AuditConsumer auditConsumer = auditConsumerResult.getResult();

        // 查询topic
        Result<Topic> topicResult = topicService.queryTopic(auditConsumer.getTid());
        if (topicResult.isNotOK()) {
            return topicResult;
        }
        // 查询cluster
        Result<Cluster> clusterResult = clusterService.queryById(topicResult.getResult().getClusterId());
        if (clusterResult.isNotOK()) {
            return Result.getResult(Status.PARAM_ERROR);
        }
        Cluster cluster = clusterResult.getResult();
        // 校验是否合法
        if (cluster.isBrokerUpdating()) {
            return Result.getResult(Status.BROKER_UPDATING);
        }

        // 构建userConsumer
        UserConsumer userConsumer = new UserConsumer();
        userConsumer.setUid(audit.getUid());
        userConsumer.setTid(auditConsumer.getTid());

        // 构建consumer
        Consumer consumer = new Consumer();
        BeanUtils.copyProperties(auditConsumer, consumer);
        consumer.setInfo(audit.getInfo());
        // 保存数据
        Result<?> saveResult = consumerService.createConsumer(cluster, consumer, userConsumer);
        if (saveResult.isNotOK()) {
            return saveResult;
        }

        // 保存限速数据
        ConsumerConfig consumerConfig = new ConsumerConfig();
        consumerConfig.setConsumer(auditConsumer.getConsumer());
        consumerConfig.setPermitsPerSecond(Double.valueOf(auditConsumer.getPermitsPerSecond()));
        consumerConfig.setEnableRateLimit(true);
        Result<?> consumerConfigResult = consumerConfigService.save(consumerConfig);
        if (consumerConfigResult.isNotOK()) {
            logger.error("save consumer{} rate limit error", auditConsumer.getConsumer());
        } else {
            // http消费需要单独设置限速
            try {
                AuditConsumerConfig auditConsumerConfig = new AuditConsumerConfig();
                auditConsumerConfig.setPermitsPerSecond(consumerConfig.getPermitsPerSecond());
                auditConsumerConfig.setEnableRateLimit(true);
                httpConsumerConfig(userInfo, consumer, auditConsumerConfig);
            } catch (Exception e) {
                logger.error("httpConsumerConfig error:{} config:{}", consumer, consumerConfig, e);
            }
        }

        // 更新申请状态
        boolean updateOK = agreeAndTip(audit, userInfo.getUser().getEmail(), consumer.getName());
        if (updateOK) {
            return Result.getOKResult();
        }
        return Result.getResult(Status.DB_UPDATE_ERR_CONSUME_OK);
    }

    /**
     * 关联生产者
     * 
     * @param aid
     * @param map
     * @return
     */
    @ResponseBody
    @RequestMapping(value = "/associateProducer", method = RequestMethod.POST)
    public Result<?> associateProducer(UserInfo userInfo,
            @RequestParam("aid") long aid) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        // 校验状态是否合法
        Audit audit = auditResult.getResult();
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }

        // 获取AuditAssociateProducer
        Result<AuditAssociateProducer> auditAssociateProducerResult = associateProducerService.query(aid);
        if (auditAssociateProducerResult.isNotOK()) {
            return auditAssociateProducerResult;
        }
        AuditAssociateProducer auditAssociateProducer = auditAssociateProducerResult.getResult();
        // 保存关联关系
        UserProducer up = new UserProducer();
        up.setTid(auditAssociateProducer.getTid());
        up.setUid(auditAssociateProducer.getUid());
        up.setProducer(auditAssociateProducer.getProducer());
        up.setProtocol(auditAssociateProducer.getProtocol());
        Result<UserProducer> userProducerResult = userProducerService.saveNoException(up);
        if (userProducerResult.isNotOK()) {
            return userProducerResult;
        }

        // 更新申请状态
        boolean updateOK = agreeAndTip(audit, userInfo.getUser().getEmail(), auditAssociateProducer.getProducer());
        if (updateOK) {
            return Result.getOKResult();
        }
        return Result.getResult(Status.DB_UPDATE_ERR_ASSOCIATE_PRODUCER_OK);
    }

    /**
     * 校验生产者链接
     * 
     * @param aid
     * @param map
     * @return
     */
    @RequestMapping("/producer/connection")
    public String validateAssociateProducer(UserInfo userInfo,
            @RequestParam("aid") long aid, Map<String, Object> map) {
        Result.setResult(map, (Object) null);
        String view = adminViewModule() + "/producerConnection";
        // 获取AuditAssociateProducer
        Result<AuditAssociateProducer> auditAssociateProducerResult = associateProducerService.query(aid);
        if (auditAssociateProducerResult.isNotOK()) {
            return view;
        }
        AuditAssociateProducer auditAssociateProducer = auditAssociateProducerResult.getResult();
        // 获取topic
        Result<Topic> topicResult = topicService.queryTopic(auditAssociateProducer.getTid());
        if (topicResult.isNotOK()) {
            return view;
        }
        Topic topic = topicResult.getResult();
        Cluster cluster = clusterService.getMQClusterById(topic.getClusterId());
        Result<ProducerConnection> producerConnectionResult = userProducerService.examineProducerConnectionInfo(
                auditAssociateProducer.getProducer(), topic.getName(), cluster, auditAssociateProducer.isProxyRemoting());

        // 组装成vo
        List<ConnectionVO> connectionVOList = new ArrayList<ConnectionVO>();
        if (producerConnectionResult.isOK()) {
            ProducerConnection producerConnection = producerConnectionResult.getResult();
            for (Connection conn : producerConnection.getConnectionSet()) {
                ConnectionVO connectionVO = new ConnectionVO();
                connectionVO.setConnection(conn);
                connectionVOList.add(connectionVO);
            }
        }
        setResult(map, connectionVOList);
        return view;
    }

    /**
     * 关联消费者
     * 
     * @param aid
     * @param map
     * @return
     */
    @ResponseBody
    @RequestMapping(value = "/associateConsumer", method = RequestMethod.POST)
    public Result<?> associateConsumer(UserInfo userInfo,
            @RequestParam("aid") long aid) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        // 校验状态是否合法
        Audit audit = auditResult.getResult();
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }

        // 获取AuditAssociateConsumer
        Result<AuditAssociateConsumer> auditAssociateConsumerResult = associateConsumerService.query(aid);
        if (auditAssociateConsumerResult.isNotOK()) {
            return auditAssociateConsumerResult;
        }
        AuditAssociateConsumer auditAssociateConsumer = auditAssociateConsumerResult.getResult();

        // 保存关联关系
        UserConsumer uc = new UserConsumer();
        uc.setTid(auditAssociateConsumer.getTid());
        uc.setUid(auditAssociateConsumer.getUid());
        uc.setConsumerId(auditAssociateConsumer.getCid());

        Result<?> userConsumerResult = userConsumerService.saveNoException(uc);
        if (userConsumerResult.isNotOK()) {
            return userConsumerResult;
        }

        // 更新申请状态
        Result<Consumer> consumerResult = consumerService.queryById(auditAssociateConsumer.getCid());
        String tip = consumerResult.isOK() ? consumerResult.getResult().getName() : null;
        boolean updateOK = agreeAndTip(audit, userInfo.getUser().getEmail(), tip);
        if (updateOK) {
            return Result.getOKResult();
        }
        return Result.getResult(Status.DB_UPDATE_ERR_ASSOCIATE_CONSUME_OK);
    }

    /**
     * 删除topic
     * 
     * @param aid
     * @param map
     * @return
     */
    @ResponseBody
    @RequestMapping(value = "/deleteTopic", method = RequestMethod.POST)
    public Result<?> deleteTopic(UserInfo userInfo,
            @RequestParam("aid") long aid) {
        // 获取audit
        Result<Audit> auditResult = auditService.queryAudit(aid);
        if (auditResult.isNotOK()) {
            return auditResult;
        }
        // 校验状态是否合法
        Audit audit = auditResult.getResult();
        if (!audit.isInitStatus()) {
            return getAuditStatusError(audit.getStatus());
        }
        // 查询 topic删除审核记录
        Result<AuditTopicDelete> auditTopicDeleteResult = auditTopicDeleteService.queryAuditTopicDelete(aid);
        if (auditTopicDeleteResult.isNotOK()) {
            return auditTopicDeleteResult;
        }
        AuditTopicDelete auditTopicDelete = auditTopicDeleteResult.getResult();
        // 查询topic记录
        Result<Topic> topicResult = topicService.queryTopic(auditTopicDelete.getTid());
        if (topicResult.isNotOK()) {
            return topicResult;
        }
        // 校验状态是否合法
        Result<Cluster> clusterResult = clusterService.queryById(topicResult.getResult().getClusterId());
        if (clusterResult.isNotOK()) {
            return Result.getResult(Status.PARAM_ERROR);
        }
        Cluster cluster = clusterResult.getResult();
        if (cluster.isBrokerUpdating()) {
            return Result.getResult(Status.BROKER_UPDATING);
        }

