package org.unidal.cat.message.storage.local;

import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.Map;

import org.codehaus.plexus.logging.LogEnabled;
import org.codehaus.plexus.logging.Logger;
import org.unidal.cat.message.storage.Bucket;
import org.unidal.cat.message.storage.BucketManager;
import org.unidal.cat.message.storage.FileBuilder;
import org.unidal.cat.message.storage.FileBuilder.FileType;
import org.unidal.cat.metric.Benchmark;
import org.unidal.cat.metric.BenchmarkManager;
import org.unidal.lookup.ContainerHolder;
import org.unidal.lookup.annotation.Inject;
import org.unidal.lookup.annotation.Named;

import com.dianping.cat.helper.TimeHelper;

@Named(type = BucketManager.class, value = "local")
public class LocalBucketManager extends ContainerHolder implements BucketManager, LogEnabled {
	@Inject
	private BenchmarkManager m_benchmarkManager;

	@Inject("local")
	private FileBuilder m_bulider;

	private Map<Integer, Map<String, Bucket>> m_buckets = new LinkedHashMap<Integer, Map<String, Bucket>>();

	private Logger m_logger;

	@Override
	public void closeBuckets(long timestamp) {
		int hour = (int) (timestamp / TimeHelper.ONE_HOUR);
		Map<String, Bucket> map = m_buckets.get(hour);

		if (map != null) {
			for (Bucket bucket : map.values()) {
				bucket.close();
				m_logger.info("close bucket " + bucket.toString());
				super.release(bucket);
			}
		}
		m_buckets.remove(hour);
	}

	private Map<String, Bucket> findOrCreateMap(Map<Integer, Map<String, Bucket>> map, int hour) {
		Map<String, Bucket> m = map.get(hour);

		if (m == null) {
			synchronized (map) {
				m = map.get(hour);

				if (m == null) {
					m = new LinkedHashMap<String, Bucket>();
					map.put(hour, m);
				}
			}
		}

		return m;
	}

	private boolean bucketExsit(String domain, String ip, int hour) {
		long timestamp = hour * 3600 * 1000L;
		Date startTime = new Date(timestamp);
		File dataPath = m_bulider.getFile(domain, startTime, ip, FileType.DATA);
		File indexPath = m_bulider.getFile(domain, startTime, ip, FileType.INDEX);

		return dataPath.exists() && indexPath.exists();
	}

	@Override
	public Bucket getBucket(String domain, String ip, int hour, boolean createIfNotExists) throws IOException {
		Map<String, Bucket> map = findOrCreateMap(m_buckets, hour);
		Bucket bucket = map.get(domain);

		if (bucket == null && createIfNotExists) {
			synchronized (map) {
				bucket = map.get(domain);

				if (bucket == null) {
					Benchmark benchmark = m_benchmarkManager.get(domain + ":" + hour);

					bucket = lookup(Bucket.class, "local");
					bucket.setBenchmark(benchmark);
					bucket.initialize(domain, ip, hour);
					map.put(domain, bucket);
				}
			}
		} else if (createIfNotExists == false) {
			if (bucketExsit(domain, ip, hour)) {
				synchronized (map) {
					bucket = map.get(domain);
					if (bucket == null) {
						bucket = lookup(Bucket.class, "local");
						bucket.initialize(domain, ip, hour);
						map.put(domain, bucket);
					}
				}
			}
		}

		return bucket;
	}

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

@@ -112,16 +112,42 @@
 l.Date;%0A
+import java.util.HashSet;%0A
 import j
@@ -191,16 +191,66 @@
 til.Map;
+%0Aimport java.util.Map.Entry;%0Aimport java.util.Set;
 %0A%0Aimport
@@ -1253,60 +1253,301 @@
 %7B%0A%09%09
-int hour = (int) (timestamp / TimeHelper.ONE_HOUR);%0A
+Set%3CInteger%3E removed = new HashSet%3CInteger%3E();%0A%0A%09%09for (Entry%3CInteger, Map%3CString, Bucket%3E%3E entry : m_buckets.entrySet()) %7B%0A%09%09%09Integer hour = entry.getKey();%0A%09%09%09long time = hour * TimeHelper.ONE_HOUR;%0A%0A%09%09%09if (time %3C= timestamp) %7B%0A%09%09%09%09removed.add(hour);%0A%09%09%09%7D%0A%09%09%7D%0A%0A%09%09for (Integer i : removed) %7B%0A%09
 %09%09Ma
@@ -1564,19 +1564,23 @@
 Bucket%3E 
-map
+buckets
  = m_buc
@@ -1588,40 +1588,19 @@
 ets.
-get(hour);%0A%0A%09%09if (map != null) %7B
+remove(i);%0A
 %0A%09%09%09
@@ -1620,19 +1620,23 @@
 ucket : 
-map
+buckets
 .values(
@@ -1656,24 +1656,51 @@
 et.close();%0A
+%09%09%09%09super.release(bucket);%0A
 %09%09%09%09m_logger
@@ -1750,66 +1750,13 @@
 %0A%09%09%09
-%09super.release(bucket);%0A%09%09%09%7D%0A%09%09%7D%0A%09%09m_buckets.remove(hour);
+%7D%0A%09%09%7D
 %0A%09%7D%0A

modify the close