package com.dianping.cat.consumer;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.codehaus.plexus.logging.LogEnabled;
import org.codehaus.plexus.logging.Logger;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
import org.unidal.helper.Threads;
import org.unidal.lookup.ContainerHolder;
import org.unidal.lookup.annotation.Inject;

import com.dianping.cat.Cat;
import com.dianping.cat.CatConstants;
import com.dianping.cat.analysis.MessageAnalyzer;
import com.dianping.cat.analysis.MessageAnalyzerManager;
import com.dianping.cat.config.black.BlackListManager;
import com.dianping.cat.message.Message;
import com.dianping.cat.message.MessageProducer;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.spi.MessageTree;
import com.dianping.cat.message.spi.core.MessageConsumer;
import com.dianping.cat.statistic.ServerStatisticManager;

public class RealtimeConsumer extends ContainerHolder implements MessageConsumer, Initializable, LogEnabled {

	@Inject
	private MessageAnalyzerManager m_analyzerManager;

	@Inject
	private ServerStatisticManager m_serverStateManager;

	@Inject
	private BlackListManager m_blackListManager;

	private Map<String, Integer> m_errorTimeDomains = new HashMap<String, Integer>();

	private PeriodManager m_periodManager;

	private long m_networkError = -1;

	private long m_black = -1;

	private Logger m_logger;

	public static final long MINUTE = 60 * 1000L;

	public static final long DURATION = 60 * MINUTE;

	@Override
	public void consume(MessageTree tree) {
		String domain = tree.getDomain();
		String ip = tree.getIpAddress();

		if (!m_blackListManager.isBlack(domain, ip)) {
			long timestamp = tree.getMessage().getTimestamp();
			Period period = m_periodManager.findPeriod(timestamp);

			if (period != null) {
				period.distribute(tree);
			} else {
				logErrorInfo(tree);
			}
		} else {
			m_black++;

			if (m_black % CatConstants.SUCCESS_COUNT == 0) {
				Cat.logEvent("Discard", domain);
			}
		}
	}

	public void doCheckpoint() throws IOException {
		MessageProducer cat = Cat.getProducer();
		Transaction t = cat.newTransaction("Checkpoint", getClass().getSimpleName());

		try {
			long currentStartTime = getCurrentStartTime();
			Period period = m_periodManager.findPeriod(currentStartTime);

			for (MessageAnalyzer analyzer : period.getAnalzyers()) {
				try {
					analyzer.doCheckpoint(false);
				} catch (Exception e) {
					Cat.logError(e);
				}
			}

			try {
				// wait dump analyzer store completed
				Thread.sleep(10 * 1000);
			} catch (InterruptedException e) {
				// ignore
			}
			t.setStatus(Message.SUCCESS);
		} catch (RuntimeException e) {
			cat.logError(e);
			t.setStatus(e);
		} finally {
			t.complete();
		}
	}

	@Override
	public void enableLogging(Logger logger) {
		m_logger = logger;
	}

	public MessageAnalyzer getCurrentAnalyzer(String name) {
		long currentStartTime = getCurrentStartTime();
		Period period = m_periodManager.findPeriod(currentStartTime);

		if (period != null) {
			return period.getAnalyzer(name);
		} else {
			return null;
		}
	}

	private long getCurrentStartTime() {
		long now = System.currentTimeMillis();
		long time = now - now % DURATION;

		return time;
	}

	public MessageAnalyzer getLastAnalyzer(String name) {
		long lastStartTime = getCurrentStartTime() - DURATION;
		Period period = m_periodManager.findPeriod(lastStartTime);

		return period == null ? null : period.getAnalyzer(name);
	}

	@Override
	public void initialize() throws InitializationException {
		m_periodManager = new PeriodManager(DURATION, m_analyzerManager, m_serverStateManager, m_logger);
		m_periodManager.init();

		Threads.forGroup("cat").start(m_periodManager);
	}

	private void logErrorInfo(MessageTree tree) {
		String domain = tree.getDomain();
		Integer size = m_errorTimeDomains.get(domain);

		if (size == null) {
			size = 1;
		} else {
			size++;
		}

		m_serverStateManager.addNetworkTimeError(1);
		m_errorTimeDomains.put(domain, size);
		m_networkError++;

		if (m_networkError % (CatConstants.ERROR_COUNT * 10) == 0) {
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd HH:mm:ss");

			m_logger.error("Error network time:" + m_errorTimeDomains);
			m_logger.error("The timestamp of message is out of range, IGNORED! "
			      + sdf.format(new Date(tree.getMessage().getTimestamp())) + " " + tree.getDomain() + " "
			      + tree.getIpAddress());
		}
	}

}
@@ -60,114 +60,8 @@
 ion;
-%0Aimport java.text.SimpleDateFormat;%0Aimport java.util.Date;%0Aimport java.util.HashMap;%0Aimport java.util.Map;
 %0A%0Aim
@@ -1259,92 +1259,8 @@
 r;%0A%0A
-%09private Map%3CString, Integer%3E m_errorTimeDomains = new HashMap%3CString, Integer%3E();%0A%0A
 %09pri
@@ -1300,44 +1300,8 @@
 r;%0A%0A
-%09private long m_networkError = -1;%0A%0A
 %09pri
@@ -1426,24 +1426,20 @@
 al long 
-DURATION
+HOUR
  = 60 * 
@@ -1807,25 +1807,50 @@
 %09%09%09%09
-logErrorInfo(tree
+m_serverStateManager.addNetworkTimeError(1
 );%0A%09
@@ -3178,24 +3178,20 @@
 - now %25 
-DURATION
+HOUR
 ;%0A%0A%09%09ret
@@ -3306,24 +3306,20 @@
 ime() - 
-DURATION
+HOUR
 ;%0A%09%09Peri
@@ -3549,16 +3549,12 @@
 ger(
-DURATION
+HOUR
 , m_
@@ -3688,714 +3688,5 @@
 %09%7D%0A%0A
-%09private void logErrorInfo(MessageTree tree) %7B%0A%09%09String domain = tree.getDomain();%0A%09%09Integer size = m_errorTimeDomains.get(domain);%0A%0A%09%09if (size == null) %7B%0A%09%09%09size = 1;%0A%09%09%7D else %7B%0A%09%09%09size++;%0A%09%09%7D%0A%0A%09%09m_serverStateManager.addNetworkTimeError(1);%0A%09%09m_errorTimeDomains.put(domain, size);%0A%09%09m_networkError++;%0A%0A%09%09if (m_networkError %25 (CatConstants.ERROR_COUNT * 10) == 0) %7B%0A%09%09%09SimpleDateFormat sdf = new SimpleDateFormat(%22yyyyMMdd HH:mm:ss%22);%0A%0A%09%09%09m_logger.error(%22Error network time:%22 + m_errorTimeDomains);%0A%09%09%09m_logger.error(%22The timestamp of message is out of range, IGNORED! %22%0A%09%09%09      + sdf.format(new Date(tree.getMessage().getTimestamp())) + %22 %22 + tree.getDomain() + %22 %22%0A%09%09%09      + tree.getIpAddress());%0A%09%09%7D%0A%09%7D%0A%0A
 %7D

modify the realtime consumer