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

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.alibaba.rocketmq.common.protocol.body.ClusterInfo;
import com.alibaba.rocketmq.common.protocol.body.KVTable;
import com.alibaba.rocketmq.common.protocol.route.BrokerData;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.cmos.msgframe.admin.common.AbstractService;
import com.cmos.msgframe.admin.service.IClusterService;
import com.cmos.msgframe.common.util.Constants;
import com.cmos.msgframe.common.util.OutPutParameter;

@Service("clusterService")
public class ClusterServiceImpl extends AbstractService implements
		IClusterService {
	static final Logger logger = LoggerFactory
			.getLogger(ClusterServiceImpl.class);

	@Override
	public OutPutParameter getClusList() throws Exception {
		DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
		OutPutParameter outData = new OutPutParameter();
		List<Map<String, String>> rsuletList = new ArrayList<Map<String, String>>();
		try {
			defaultMQAdminExt.start();
			ClusterInfo clusterInfoSerializeWrapper = defaultMQAdminExt.examineBrokerClusterInfo();
			Iterator<Map.Entry<String, Set<String>>> itCluster = clusterInfoSerializeWrapper
					.getClusterAddrTable().entrySet().iterator();
			 while (itCluster.hasNext()){
				 Map.Entry<String, Set<String>> next = itCluster.next(); 
				 String clusterName = next.getKey();
				 Set<String> brokerNameSet = new HashSet<String>();
		         brokerNameSet.addAll(next.getValue());
		         for (String brokerName : brokerNameSet){
		        	 BrokerData brokerData = clusterInfoSerializeWrapper.getBrokerAddrTable().get(brokerName);
		                if (brokerData != null){
		                	Set<Map.Entry<Long, String>> brokerAddrSet = brokerData
		                            .getBrokerAddrs().entrySet();
		                    Iterator<Map.Entry<Long, String>> itAddr = brokerAddrSet
		                            .iterator();
		                    while (itAddr.hasNext()){
		                    	Map<String, String> map = new HashMap<String, String>();
		                    	map.put("clusterName", clusterName);
		                    	map.put("brokerName", brokerName);
		                    	map.put("brokerName", brokerName);
		                    	
		                    	Map.Entry<Long, String> next1 = itAddr.next();
		                    	map.put("brokerID", String.valueOf(next1.getKey()));
		                    	map.put("brokerAddr", String.valueOf(next1.getValue()));
		                    	 try{
		                             KVTable kvTable = defaultMQAdminExt.fetchBrokerRuntimeStats(next1.getValue());
		                           	map.put("version",  kvTable.getTable().get("brokerVersionDesc"));
		                            String putTps = kvTable.getTable().get("putTps");
		                            String getTransferedTps = kvTable.getTable().get("getTransferedTps");
		                            {
		                                String[] tpss = putTps.split(" ");
		                                if (tpss != null && tpss.length > 0)
		                                {
		                                	  double in = Double.parseDouble(tpss[0]);
	                                	  	  map.put("inTps",  String.valueOf(in));
		                                }
		                            }
		                            {
		                                String[] tpss = getTransferedTps.split(" ");
		                                if (tpss != null && tpss.length > 0)
		                                {
		                                	double out = Double.parseDouble(tpss[0]);
		                                  	map.put("outTps",  String.valueOf(out));
		                                }
		                            }
		                            String msgPutTotalYesterdayMorning = kvTable
		                                    .getTable().get(
		                                            "msgPutTotalYesterdayMorning");
		                            String msgPutTotalTodayMorning = kvTable.getTable()
		                                    .get("msgPutTotalTodayMorning");
		                            String msgPutTotalTodayNow = kvTable.getTable()
		                                    .get("msgPutTotalTodayNow");
		                            String msgGetTotalYesterdayMorning = kvTable
		                                    .getTable().get(
		                                            "msgGetTotalYesterdayMorning");
		                            String msgGetTotalTodayMorning = kvTable.getTable()
		                                    .get("msgGetTotalTodayMorning");
		                            String msgGetTotalTodayNow = kvTable.getTable()
		                                    .get("msgGetTotalTodayNow");
		                           long InTotalYest = Long
		                                    .parseLong(msgPutTotalTodayMorning)
		                                    - Long.parseLong(msgPutTotalYesterdayMorning);
		                           long OutTotalYest = Long
		                                    .parseLong(msgGetTotalTodayMorning)
		                                    - Long.parseLong(msgGetTotalYesterdayMorning);

		                           long InTotalToday = Long.parseLong(msgPutTotalTodayNow)
		                                    - Long.parseLong(msgPutTotalTodayMorning);
		                           long OutTotalToday = Long.parseLong(msgGetTotalTodayNow)
		                                    - Long.parseLong(msgGetTotalTodayMorning);
		                           map.put("InTotalYest", String.valueOf(InTotalYest));
		                           map.put("OutTotalYest", String.valueOf(OutTotalYest));
		                           map.put("InTotalToday", String.valueOf(InTotalToday));
		                           map.put("OutTotalToday", String.valueOf(OutTotalToday));
		                           map.put("InTotalNow", msgPutTotalTodayNow);
		                           map.put("OutTotalNow", msgGetTotalTodayNow);
		                           rsuletList.add(map);
		                         }catch (Exception e){
		                             logger.error(e.getMessage(), e);
		                         }
		                    }
		                }
		         }
			 }
			 outData.setBeans(rsuletList);
			 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);
		}
	}

}