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

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

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

import com.alibaba.rocketmq.client.QueryResult;
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.alibaba.rocketmq.tools.admin.DefaultMQAdminExt;
import com.cmos.msgframe.admin.common.AbstractService;
import com.cmos.msgframe.admin.service.IMessageService;
import com.cmos.msgframe.common.util.Constants;
import com.cmos.msgframe.common.util.OutPutParameter;
@Service("messageService")
public class MessageServiceImpl extends AbstractService implements IMessageService{
	static final Logger logger = LoggerFactory
			.getLogger(MessageServiceImpl.class);
	@Override
	public OutPutParameter getMessageById(String msgId) throws Exception {
		Map<String, Object> map = new LinkedHashMap<String, Object>();
        DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
        OutPutParameter outData = new OutPutParameter();
        try {
        	if(StringUtils.isEmpty(msgId)||msgId.equals("null")){
        		outData.setReturnCode(Constants.SYSTEM_ERROR);
        		outData.setReturnMessage("消息ID不能为空");
        		return outData;
        	}
            defaultMQAdminExt.start();
            MessageExt msg = defaultMQAdminExt.viewMessage(msgId);
            String bodyTmpFilePath = createBodyFile(msg);
            map.put("msgBody",new String(msg.getBody(),"utf-8"));
            map.put("Topic", msg.getTopic());
            map.put("Tags", msg.getTags());
            map.put("Keys", msg.getKeys());
            map.put("Queue ID", String.valueOf(msg.getQueueId()));
            map.put("Queue Offset:", String.valueOf(msg.getQueueOffset()));
            map.put("CommitLog Offset:", String.valueOf(msg.getCommitLogOffset()));
            map.put("Born Timestamp:", UtilAll.timeMillisToHumanString2(msg.getBornTimestamp()));
            map.put("Store Timestamp:", UtilAll.timeMillisToHumanString2(msg.getStoreTimestamp()));
            map.put("Born Host:", RemotingHelper.parseSocketAddressAddr(msg.getBornHost()));
            map.put("Store Host:", RemotingHelper.parseSocketAddressAddr(msg.getStoreHost()));
            map.put("System Flag:", String.valueOf(msg.getSysFlag()));
            map.put("Properties:", msg.getProperties() != null ? msg.getProperties().toString() : "");
            map.put("Message Body Path:", bodyTmpFilePath);
            outData.setBean(map);
            outData.setReturnCode(Constants.IS_OK);
            return outData;
        }catch (Exception e) {
           outData.setReturnCode(Constants.SYSTEM_ERROR);
		   outData.setReturnMessage("根据消息ID查询消息出现异常");
		   logger.error("根据消息ID查询消息出现异常",e);
		   return outData;
        }
        finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
	}

	@Override
	public OutPutParameter getMessageByKey(String topic, String Key,int start,int endIndex)
			throws Exception {
		 DefaultMQAdminExt defaultMQAdminExt = getDefaultMQAdminExt();
		 List<Map<String, String>> resultList = new ArrayList<Map<String,String>>();
		 OutPutParameter outData = new OutPutParameter();
		 try {
			 if(StringUtils.isEmpty(topic)||topic.equals("null")){
	        		outData.setReturnCode(Constants.SYSTEM_ERROR);
	        		outData.setReturnMessage("主题不能为空");
	        		return outData;
        	 }
			 if(StringUtils.isEmpty(Key)||Key.equals("null")){
	        		outData.setReturnCode(Constants.SYSTEM_ERROR);
	        		outData.setReturnMessage("key值不能为空");
	        		return outData;
			 }
			 defaultMQAdminExt.start();
			 long end = System.currentTimeMillis();
	            long begin = end - (6 * 60 * 60 * 1000);
	            QueryResult queryResult = defaultMQAdminExt.queryMessage(topic, Key, 100, begin, end); 
	            for (MessageExt msg : queryResult.getMessageList()) {
	            	Map<String, String> map = new HashMap<String, String>();
	            	msg.getProperties().put("UNIQ_KEY", null);
	            	map.put("msgId", msg.getMsgId());
	            	map.put("queueId", String.valueOf(msg.getQueueId()));
	            	map.put("queueOffset", String.valueOf(msg.getQueueOffset()));
	            	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, endIndex));
	            outData.setReturnCode(Constants.IS_OK);
	            return outData;
		} catch (Exception e) {
		   outData.setReturnCode(Constants.SYSTEM_ERROR);
		   outData.setReturnMessage("根据消息Key查询消息出现异常");
		   logger.error("根据消息Key查询消息出现异常",e);
		   return outData;
		}finally {
            shutdownDefaultMQAdminExt(defaultMQAdminExt);
        }
	}

	@Override
	public OutPutParameter getMessageByOffset(String topic,
			String brokerName, String offset, String queueId) throws Exception {
		DefaultMQPullConsumer defaultMQPullConsumer = new DefaultMQPullConsumer(MixAll.TOOLS_CONSUMER_GROUP);
		defaultMQPullConsumer.setInstanceName(Long.toString(System.currentTimeMillis()));
		OutPutParameter outData = new OutPutParameter();
		try {
			 if(StringUtils.isEmpty(topic)||topic.equals("null")){
	        		outData.setReturnCode(Constants.SYSTEM_ERROR);
	        		outData.setReturnMessage("主题不能为空");
	        		return outData;
			 }
			 if(StringUtils.isEmpty(brokerName)||brokerName.equals("null")){
	        		outData.setReturnCode(Constants.SYSTEM_ERROR);
	        		outData.setReturnMessage("BrokerName不能为空");
	        		return outData;
			 }
			 if(StringUtils.isEmpty(offset)||offset.equals("null")){
	        		outData.setReturnCode(Constants.SYSTEM_ERROR);
	        		outData.setReturnMessage("Offset值不能为空");
	        		return outData;
			 }
			 if(StringUtils.isEmpty(queueId)||queueId.equals("null")){
				 outData.setReturnCode(Constants.SYSTEM_ERROR);
				 outData.setReturnMessage("key值不能为空");
				 return outData;
			 }
			MessageQueue mq = new MessageQueue();
			mq.setTopic(topic);
			mq.setBrokerName(brokerName);
			mq.setQueueId(Integer.parseInt(queueId));
			defaultMQPullConsumer.start();
			PullResult pullResult = defaultMQPullConsumer.pull(mq, "*", Long.parseLong(offset), 1);
			 if (pullResult != null) {
				 switch (pullResult.getPullStatus()) {
			     case FOUND:
			    	 pullResult.getMsgFoundList().get(0).getProperties().put("UNIQ_KEY", null);
//			    	 System.out.println(pullResult.getMsgFoundList().get(0).getMsgId());
			    	 outData= getMessageById(pullResult.getMsgFoundList().get(0).getMsgId());
			         return outData;
			     case NO_MATCHED_MSG:
			     case NO_NEW_MSG:
			     case OFFSET_ILLEGAL:
			     default:
			         break;
			     }
			 }
		} catch (Exception e) {
			   outData.setReturnCode(Constants.SYSTEM_ERROR);
			   outData.setReturnMessage("通过offset查询消息失败");
			   logger.error("通过offset查询消息失败",e);
			   return outData;
		}finally {
			defaultMQPullConsumer.shutdown();
        }
		return outData;
	}

	private String createBodyFile(MessageExt msg) throws IOException {
        DataOutputStream dos = null;

        try {
            String bodyTmpFilePath = "/tmp/rocketmq/msgbodys";
            File file = new File(bodyTmpFilePath);
            if (!file.exists()) {
                file.mkdirs();
            }
            bodyTmpFilePath = bodyTmpFilePath + "/" + msg.getMsgId();
            dos = new DataOutputStream(new FileOutputStream(bodyTmpFilePath));
            dos.write(msg.getBody());
            return bodyTmpFilePath;
        }
        finally {
            if (dos != null)
                dos.close();
        }
    }
}