package com.cmos.msgframe.admin.service.Impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.protocol.body.Connection;
import com.alibaba.rocketmq.common.protocol.body.ConsumerConnection;
import com.alibaba.rocketmq.common.protocol.body.GroupList;
import com.alibaba.rocketmq.common.protocol.body.ProducerConnection;
import com.alibaba.rocketmq.common.protocol.body.TopicList;
import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.cmos.msgframe.admin.common.AbstractService;
import com.cmos.msgframe.admin.service.IConnectionService;
import com.cmos.msgframe.common.util.Constants;
import com.cmos.msgframe.common.util.OutPutParameter;

@Service("connectionService")
public class ConnectionServiceImpl extends AbstractService implements
		IConnectionService {
	static final Logger logger = LoggerFactory
			.getLogger(ConnectionServiceImpl.class);
	@Override
	public OutPutParameter getConConnectionList(String consumerGroup,
			String topic,int start,int end) throws Exception {
		DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
		List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
		OutPutParameter outData = new OutPutParameter();
		try {
			if (StringUtils.isNotBlank(consumerGroup)&&!consumerGroup.equals("null")) {
				defaultMQAdminExt.start();
				Map<String, String> map = getConConnectionInfo(consumerGroup,
						defaultMQAdminExt);
				if (null != map) {
					resultList.add(map);
				}
			} else if (StringUtils.isNotBlank(topic)&&!topic.equals("null")) {
				defaultMQAdminExt.start();
				GroupList groupList = defaultMQAdminExt
						.queryTopicConsumeByWho(topic);
				for (String consumerGroupTemp : groupList.getGroupList()) {
					Map<String, String> map = getConConnectionInfo(
							consumerGroupTemp, defaultMQAdminExt);
					if (null != map) {
						resultList.add(map);
					}
				}
			} else {
				defaultMQAdminExt.start();
				TopicList topicList = defaultMQAdminExt.fetchAllTopicList();
				for (String topicTemp : topicList.getTopicList()) {
					if (topicTemp.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
						continue;
					}
					GroupList groupList = defaultMQAdminExt.queryTopicConsumeByWho(topicTemp);
					for (String consumerGroupTemp : groupList.getGroupList()) {
						Map<String, String> map = getConConnectionInfo(consumerGroupTemp, defaultMQAdminExt);
						if (null != map) {
							resultList.add(map);
						}
					}
				}
			}
			Map<String, Object> totalMap = new HashMap<String, Object>();
			totalMap.put("total", resultList.size());
			outData.setBean(totalMap);
			outData.setBeans((List<Map<String, String>>) page(resultList, start, end));
			outData.setReturnCode(Constants.IS_OK);
			return outData;
		} catch (Exception e) {
			outData.setReturnCode(Constants.SYSTEM_ERROR);
		    outData.setReturnMessage("查询消费者连接信息出现异常,请检查消费组查询条件是否正确");
		    logger.error("查询消费者连接信息出现异常,请检查消费组查询条件是否正确",e);
		    return outData;
		} finally {
			shutdownDefaultMQAdminExt(defaultMQAdminExt);
		}
	}

	private Map<String, String> getConConnectionInfo(String consumerGroup,
			DefaultMQAdminExt defaultMQAdminExt)  throws Exception {
		Map<String, String> map = new HashMap<String, String>();
		try {
			ConsumerConnection cc = null;
			try {
				cc = defaultMQAdminExt.examineConsumerConnectionInfo(consumerGroup);
			} catch (Exception e) {
				return null;
			}
			map.put("consumerGroup", consumerGroup);
			map.put("consumerCount",String.valueOf(cc.getConnectionSet().size()));
			map.put("consumeFromWhere",String.valueOf(cc.getConsumeFromWhere()));
			map.put("messageModel", cc.getMessageModel().toString());
			map.put("consumeType", cc.getConsumeType().toString());
			for (Entry<String, SubscriptionData> entry : cc
					.getSubscriptionTable().entrySet()) {
				if (entry.getKey().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
					continue;
				}
				map.put("topic", entry.getKey());
				break;
			}
		} catch (Exception e) {

		}
		return map;
	}

	@Override
	public OutPutParameter getConConnectiondetail(
			String consumerGroup, String topic,int start,int end) throws Exception {
		DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
		List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
		OutPutParameter outData = new OutPutParameter();
		ConsumerConnection cc = null;
		try {
			if(StringUtils.isEmpty(consumerGroup)){
				outData.setReturnCode(Constants.SYSTEM_ERROR);
				outData.setReturnMessage("消费组不能为空");
				return outData;
			}
			if(StringUtils.isEmpty(topic)){
				outData.setReturnCode(Constants.SYSTEM_ERROR);
				outData.setReturnMessage("主题不能为空");
				return outData;
			}
			defaultMQAdminExt.start();
			cc = defaultMQAdminExt.examineConsumerConnectionInfo(consumerGroup);
			for (Connection conn : cc.getConnectionSet()) {
				Map<String, String> map = new HashMap<String, String>();
				map.put("clientId", conn.getClientId());
				map.put("topicName", topic);
				map.put("clientAddr", conn.getClientAddr());
				map.put("version", String.valueOf(conn.getVersion()));
				map.put("language", conn.getLanguage().toString());
				resultList.add(map);
			}
			Map<String, Object> totalMap = new HashMap<String, Object>();
			totalMap.put("total", resultList.size());
			outData.setBean(totalMap);
			outData.setBeans((List<Map<String, String>>) page(resultList, start, end));
			outData.setReturnCode(Constants.IS_OK);
			return outData;
		} catch (Exception e) {
			outData.setReturnCode(Constants.SYSTEM_ERROR);
		    outData.setReturnMessage("查询消费者连接信息出现异常");
		    logger.error("查询消费者连接信息出现异常",e);
		    return outData;
		} finally {
			shutdownDefaultMQAdminExt(defaultMQAdminExt);
		}
	}

	@Override
	public OutPutParameter getProConnectionList(String producerGroup,
			String topic) throws Exception {
		DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
		List<Map<String, String>> resultList = new ArrayList<Map<String, String>>();
		OutPutParameter outData = new OutPutParameter();
		try {
			if (StringUtils.isBlank(producerGroup)||producerGroup.equals("null")) {
				outData.setBeans(resultList);
				outData.setReturnCode(Constants.SYSTEM_ERROR);
				outData.setReturnMessage("生产者组不能为空");
				return outData;
			}
			if (StringUtils.isBlank(topic)||producerGroup.equals("null")) {
				outData.setBeans(resultList);
				outData.setReturnCode(Constants.SYSTEM_ERROR);
				outData.setReturnMessage("主题不能为空");
				return outData;
			}
			defaultMQAdminExt.start();
			ProducerConnection pc = defaultMQAdminExt
					.examineProducerConnectionInfo(producerGroup, topic);
			for (Connection conn : pc.getConnectionSet()) {
				Map<String, String> map = new HashMap<String, String>();
				map.put("clientId", conn.getClientId());
				map.put("clientAddr", conn.getClientAddr());
				map.put("topicName", topic);
				map.put("version", String.valueOf(conn.getVersion()));
				map.put("language", conn.getLanguage().toString());
				resultList.add(map);
			}
			Map<String, Object> totalMap = new HashMap<String, Object>();
			totalMap.put("total", resultList.size());
			outData.setBean(totalMap);
			outData.setBeans(resultList);
			outData.setReturnCode(Constants.IS_OK);
			return outData;
		} catch (Exception e) {
			outData.setReturnCode(Constants.SYSTEM_ERROR);
		    outData.setReturnMessage("查询生产者连接信息出现异常");
		    logger.error("查询生产者连接信息出现异常",e);
		    return outData;
		} finally {
			shutdownDefaultMQAdminExt(defaultMQAdminExt);
		}
	}

}