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

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.unidal.cat.message.storage.Block;
import org.unidal.cat.message.storage.BlockWriter;
import org.unidal.cat.message.storage.Bucket;
import org.unidal.cat.message.storage.BucketManager;
import org.unidal.cat.metric.Benchmark;
import org.unidal.cat.metric.BenchmarkEnabled;
import org.unidal.cat.metric.BenchmarkManager;
import org.unidal.cat.metric.Metric;
import org.unidal.lookup.annotation.Inject;
import org.unidal.lookup.annotation.Named;

import com.dianping.cat.configuration.NetworkInterfaceManager;

@Named(type = BlockWriter.class, instantiationStrategy = Named.PER_LOOKUP)
public class DefaultBlockWriter implements BlockWriter {
	@Inject
	private BucketManager m_bucketManager;

	@Inject
	private BenchmarkManager m_benchmarkManager;

	private int m_index;

	private BlockingQueue<Block> m_queue;

	private AtomicBoolean m_enabled;

	private CountDownLatch m_latch;

	private long m_timestamp;

	@Override
	public String getName() {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:ss");

		return getClass().getSimpleName() + " " + sdf.format(new Date(m_timestamp)) + "-" + m_index;
	}

	@Override
	public void initialize(long timestamp, int index, BlockingQueue<Block> queue) {
		m_timestamp = timestamp;
		m_index = index;
		m_queue = queue;
		m_enabled = new AtomicBoolean(true);
		m_latch = new CountDownLatch(1);
	}

	@Override
	public void run() {
		String ip = NetworkInterfaceManager.INSTANCE.getLocalHostAddress();
		Benchmark benchmark = m_benchmarkManager.get("BlockWriter-" + m_index);
		Metric metric = benchmark.get("wait");
		Block block;

		try {
			while (m_enabled.get()) {
				metric.start();
				block = m_queue.poll(5, TimeUnit.MILLISECONDS);
				metric.end();

				if (block != null) {
					try {
						Bucket bucket = m_bucketManager.getBucket(block.getDomain(), ip, block.getHour(), true);

						if (bucket instanceof BenchmarkEnabled) {
							((BenchmarkEnabled) bucket).setBenchmark(benchmark);
						}

						bucket.puts(block.getData(), block.getMappings());
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		} catch (InterruptedException e) {
			// ignore it
		}

		System.out.println(getClass().getSimpleName() + "-" + m_index + " is shutdown");
		m_latch.countDown();
		benchmark.print();
	}

	@Override
	public void shutdown() {
		m_enabled.set(false);

		try {
			m_latch.await();
		} catch (InterruptedException e) {
			// ignore it
		}
	}
}

@@ -728,16 +728,45 @@
 Named;%0A%0A
+import com.dianping.cat.Cat;%0A
 import c
@@ -819,16 +819,61 @@
 Manager;
+%0Aimport com.dianping.cat.message.Transaction;
 %0A%0A@Named
@@ -1264,16 +1264,39 @@
 stamp;%0A%0A
+%09private int m_count;%0A%0A
 %09@Overri
@@ -1982,23 +1982,12 @@
 le (
-m_enabled.get()
+true
 ) %7B%0A
@@ -2336,121 +2336,424 @@
 %09%09%09%09
-bucket.puts(block.getData(), block.getMappings());%0A%09%09%09%09%09%7D catch (Exception e) %7B%0A%09%09%09%09%09%09e.printStackTrace();%0A%09%09%09%09%09%7D
+if ((++m_count) %25 100 == 0) %7B%0A%09%09%09%09%09%09%09Transaction t = Cat.newTransaction(%22Block%22, block.getDomain());%0A%0A%09%09%09%09%09%09%09bucket.puts(block.getData(), block.getMappings());%0A%09%09%09%09%09%09%09t.setStatus(Transaction.SUCCESS);%0A%09%09%09%09%09%09%09t.complete();%0A%09%09%09%09%09%09%7D else %7B%0A%09%09%09%09%09%09%09bucket.puts(block.getData(), block.getMappings());%0A%09%09%09%09%09%09%7D%0A%09%09%09%09%09%7D catch (Exception e) %7B%0A%09%09%09%09%09%09Cat.logError(e);%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D else if (m_enabled.get() == false) %7B%0A%09%09%09%09%09break;
 %0A%09%09%09

fix DefaultBlockWriter thead end , if queue have some block bug