package com.dianping.cat.analysis;

import org.codehaus.plexus.logging.Logger;
import org.unidal.lookup.ContainerHolder;
import org.unidal.lookup.annotation.Inject;

import com.dianping.cat.Cat;
import com.dianping.cat.config.server.ServerConfigManager;
import com.dianping.cat.message.spi.MessageQueue;
import com.dianping.cat.message.spi.MessageTree;
import com.dianping.cat.report.ReportManager;

public abstract class AbstractMessageAnalyzer<R> extends ContainerHolder implements MessageAnalyzer {
	public static final long MINUTE = 60 * 1000L;

	public static final long ONE_HOUR = 60 * 60 * 1000L;

	public static final long ONE_DAY = 24 * ONE_HOUR;

	@Inject
	protected ServerConfigManager m_serverConfigManager;

	protected long m_extraTime;

	protected long m_startTime;

	protected long m_duration;

	protected Logger m_logger;

	private long m_errors = 0;

	private volatile boolean m_active = true;

	protected int m_index;

	@Override
	public void analyze(MessageQueue queue) {
		while (!isTimeout() && isActive()) {
			MessageTree tree = queue.poll();

			if (tree != null) {
				try {
					process(tree);
				} catch (Throwable e) {
					m_errors++;

					if (m_errors == 1 || m_errors % 10000 == 0) {
						Cat.logError(e);
					}
				}
			}
		}

		while (true) {
			MessageTree tree = queue.poll();

			if (tree != null) {
				try {
					process(tree);
				} catch (Throwable e) {
					m_errors++;

					if (m_errors == 1 || m_errors % 10000 == 0) {
						Cat.logError(e);
					}
				}
			} else {
				break;
			}
		}
	}

	@Override
	public void destroy() {
		super.release(this);
		ReportManager<?> manager = this.getReportManager();

		if (manager != null) {
			manager.destory();
		}
	}

	@Override
	public abstract void doCheckpoint(boolean atEnd);

	@Override
	public int getAnanlyzerCount() {
		return 1;
	}

	protected long getExtraTime() {
		return m_extraTime;
	}

	public abstract R getReport(String domain);

	@Override
	public long getStartTime() {
		return m_startTime;
	}

	@Override
	public void initialize(long startTime, long duration, long extraTime) {
		m_extraTime = extraTime;
		m_startTime = startTime;
		m_duration = duration;

		loadReports();
	}

	protected boolean isActive() {
		synchronized (this) {
			return m_active;
		}
	}

	protected boolean isLocalMode() {
		return m_serverConfigManager.isLocalMode();
	}

	protected boolean isTimeout() {
		long currentTime = System.currentTimeMillis();
		long endTime = m_startTime + m_duration + m_extraTime;

		return currentTime > endTime;
	}

	protected abstract void loadReports();

	protected abstract void process(MessageTree tree);

	public void shutdown() {
		synchronized (this) {
			m_active = false;
		}
	}

	public void setIndex(int index) {
		m_index = index;
	}

}

@@ -27,24 +27,77 @@
 nalysis;%0D%0A%0D%0A
+import java.util.concurrent.atomic.AtomicBoolean;%0D%0A%0D%0A
 import org.c
@@ -124,24 +124,24 @@
 ng.Logger;%0D%0A
-
 import org.u
@@ -959,18 +959,15 @@
 ate 
-volatile b
+AtomicB
 oole
@@ -980,20 +980,39 @@
 ctive = 
+new AtomicBoolean(
 true
+)
 ;%0D%0A%0D%0A%09pr
@@ -2399,34 +2399,8 @@
  %7B%0D%0A
-%09%09synchronized (this) %7B%0D%0A%09
 %09%09re
@@ -2412,22 +2412,23 @@
 m_active
-;%0D%0A%09%09%7D
+.get();
 %0D%0A%09%7D%0D%0A%0D%0A
@@ -2801,93 +2801,8 @@
 %0D%0A%0D%0A
-%09public void shutdown() %7B%0D%0A%09%09synchronized (this) %7B%0D%0A%09%09%09m_active = false;%0D%0A%09%09%7D%0D%0A%09%7D%0D%0A%0D%0A
 %09pub
@@ -2859,11 +2859,66 @@
 %0D%0A%09%7D%0D%0A%0D%0A
+%09public void shutdown() %7B%0D%0A%09%09m_active.set(false);%0D%0A%09%7D%0D%0A
 %7D%0D%0A

fix blocking issue when shutting down server