package com.ctrip.hermes.producer.monitor;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.unidal.lookup.annotation.Named;
import org.unidal.tuple.Pair;

import com.ctrip.hermes.core.constants.CatConstants;
import com.ctrip.hermes.core.message.ProducerMessage;
import com.ctrip.hermes.core.transport.command.SendMessageCommand;
import com.ctrip.hermes.core.transport.command.SendMessageResultCommand;
import com.ctrip.hermes.producer.status.ProducerStatusMonitor;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.internal.DefaultTransaction;
import com.dianping.cat.message.spi.MessageTree;
import com.google.common.util.concurrent.SettableFuture;

/**
 * @author Leo Liang(jhliang@ctrip.com)
 *
 */
@Named(type = SendMessageResultMonitor.class)
public class DefaultSendMessageResultMonitor implements SendMessageResultMonitor {
	private static final Logger log = LoggerFactory.getLogger(DefaultSendMessageResultMonitor.class);

	private Map<Long, Pair<SendMessageCommand, SettableFuture<Void>>> m_cmds = new ConcurrentHashMap<>();

	@Override
	public Future<Void> monitor(SendMessageCommand cmd) {
		SettableFuture<Void> future = SettableFuture.create();
		m_cmds.put(cmd.getHeader().getCorrelationId(), new Pair<SendMessageCommand, SettableFuture<Void>>(cmd, future));
		return future;
	}

	@Override
	public void resultReceived(SendMessageResultCommand result) {
		if (result != null) {
			Pair<SendMessageCommand, SettableFuture<Void>> pair = m_cmds.remove(result.getHeader().getCorrelationId());
			if (pair != null) {
				try {
					SendMessageCommand sendMessageCommand = pair.getKey();
					SettableFuture<Void> future = pair.getValue();

					ProducerStatusMonitor.INSTANCE.brokerResultReceived(sendMessageCommand.getTopic(),
					      sendMessageCommand.getPartition(), sendMessageCommand.getMessageCount());

					future.set(null);
					sendMessageCommand.onResultReceived(result);
					tracking(sendMessageCommand, true);
				} catch (Exception e) {
					log.warn("Exception occurred while calling resultReceived", e);
				}

			}
		}
	}

	private void tracking(SendMessageCommand sendMessageCommand, boolean success) {
		String status = success ? Transaction.SUCCESS : "Timeout";

		for (List<ProducerMessage<?>> msgs : sendMessageCommand.getProducerMessages()) {
			for (ProducerMessage<?> msg : msgs) {
				Transaction t = Cat.newTransaction("Message.Produce.Acked", msg.getTopic());
				MessageTree tree = Cat.getManager().getThreadLocalMessageTree();

				String msgId = msg.getDurableSysProperty(CatConstants.SERVER_MESSAGE_ID);
				String parentMsgId = msg.getDurableSysProperty(CatConstants.CURRENT_MESSAGE_ID);
				String rootMsgId = msg.getDurableSysProperty(CatConstants.ROOT_MESSAGE_ID);

				tree.setMessageId(msgId);
				tree.setParentMessageId(parentMsgId);
				tree.setRootMessageId(rootMsgId);

				Transaction elapseT = Cat.newTransaction("Message.Produce.Elapse", msg.getTopic());
				if (elapseT instanceof DefaultTransaction) {
					((DefaultTransaction) elapseT).setDurationStart(msg.getBornTimeNano());
					elapseT.addData("command.message.count", sendMessageCommand.getMessageCount());
				}
				elapseT.setStatus(status);
				elapseT.complete();

				t.setStatus(status);
				t.complete();
			}
		}
	}

	@Override
	public void cancel(SendMessageCommand cmd) {
		m_cmds.remove(cmd.getHeader().getCorrelationId());
	}

}

@@ -164,16 +164,65 @@
 .Future;
+%0Aimport java.util.concurrent.locks.ReentrantLock;
 %0A%0Aimport
@@ -1282,16 +1282,70 @@
 p%3C%3E();%0A%0A
+%09private ReentrantLock m_lock = new ReentrantLock();%0A%0A
 %09@Overri
@@ -1402,16 +1402,42 @@
  cmd) %7B%0A
+%09%09m_lock.lock();%0A%09%09try %7B%0A%09
 %09%09Settab
@@ -1485,16 +1485,17 @@
 eate();%0A
+%09
 %09%09m_cmds
@@ -1603,16 +1603,17 @@
 re));%0A%09%09
+%09
 return f
@@ -1619,16 +1619,54 @@
 future;%0A
+%09%09%7D finally %7B%0A%09%09%09m_lock.unlock();%0A%09%09%7D%0A
 %09%7D%0A%0A%09@Ov
@@ -1815,16 +1815,60 @@
 %3E pair =
+ null;%0A%09%09%09m_lock.lock();%0A%09%09%09try %7B%0A%09%09%09%09pair =
  m_cmds.
@@ -1914,16 +1914,57 @@
 nId());%0A
+%09%09%09%7D finally %7B%0A%09%09%09%09m_lock.unlock();%0A%09%09%09%7D%0A
 %09%09%09if (p
@@ -3749,24 +3749,50 @@
 mand cmd) %7B%0A
+%09%09m_lock.lock();%0A%09%09try %7B%0A%09
 %09%09m_cmds.rem
@@ -3832,14 +3832,52 @@
 nId());%0A
+%09%09%7D finally %7B%0A%09%09%09m_lock.unlock();%0A%09%09%7D%0A
 %09%7D%0A%0A%7D%0A

add lock in DefaultSendMessageResultMonitor