package com.dianping.cat.report.page.model.spi.internal;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
import org.unidal.helper.Splitters;
import org.unidal.helper.Threads;
import org.unidal.lookup.annotation.Inject;

import com.dianping.cat.Cat;
import com.dianping.cat.configuration.ServerConfigManager;
import com.dianping.cat.message.Event;
import com.dianping.cat.message.Message;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.report.page.model.spi.ModelService;
import com.dianping.cat.service.ModelRequest;
import com.dianping.cat.service.ModelResponse;

public abstract class BaseCompositeModelService<T> extends ModelServiceWithCalSupport implements ModelService<T>,
      Initializable {
	private static ExecutorService s_threadPool = Threads.forPool().getFixedThreadPool("Cat-ModelService", 50);

	// introduce another list is due to a bug inside Plexus ComponentList
	private List<ModelService<T>> m_allServices = new ArrayList<ModelService<T>>();

	@Inject
	protected ServerConfigManager m_configManager;

	private String m_name;

	@Inject
	private List<ModelService<T>> m_services;

	public BaseCompositeModelService(String name) {
		m_name = name;
	}

	protected abstract BaseRemoteModelService<T> createRemoteService();

	@Override
	public String getName() {
		return m_name;
	}

	@Override
	public void initialize() throws InitializationException {
		m_allServices.addAll(m_services);

		String remoteServers = m_configManager.getConsoleRemoteServers();
		List<String> endpoints = Splitters.by(',').noEmptyItem().trim().split(remoteServers);

		for (String endpoint : endpoints) {
			int pos = endpoint.indexOf(':');
			String host = (pos > 0 ? endpoint.substring(0, pos) : endpoint);
			int port = (pos > 0 ? Integer.parseInt(endpoint.substring(pos + 1)) : 2281);
			BaseRemoteModelService<T> remote = createRemoteService();

			remote.setHost(host);
			remote.setPort(port);
			m_allServices.add(remote);
		}
	}

	@Override
	public ModelResponse<T> invoke(final ModelRequest request) {
		int size = m_allServices.size();
		final List<ModelResponse<T>> responses = Collections.synchronizedList(new ArrayList<ModelResponse<T>>(size));
		final Semaphore semaphore = new Semaphore(0);
		final Transaction t = Cat.getProducer().newTransaction("ModelService", getClass().getSimpleName());
		int count = 0;

		t.setStatus(Message.SUCCESS);
		t.addData("request", request);
		t.addData("thread", Thread.currentThread());

		for (final ModelService<T> service : m_allServices) {
			if (!service.isEligable(request)) {
				continue;
			}

			// save current transaction so that child thread can access it
			if (service instanceof ModelServiceWithCalSupport) {
				((ModelServiceWithCalSupport) service).setParentTransaction(t);
			}

			s_threadPool.submit(new Runnable() {
				@Override
				public void run() {
					try {
						ModelResponse<T> response = service.invoke(request);

						if (response.getException() != null) {
							logError(response.getException());
						}
						if (response != null && response.getModel() != null) {
							responses.add(response);
						}
					} catch (Exception e) {
						logError(e);
						t.setStatus(e);
					} finally {
						semaphore.release();
					}
				}
			});

			count++;
		}

		try {
			semaphore.tryAcquire(count, 10000, TimeUnit.MILLISECONDS); // 10 seconds timeout
		} catch (InterruptedException e) {
			// ignore it
			t.setStatus(e);
		} finally {
			t.complete();
		}

		String requireAll = request.getProperty("requireAll");

		if (requireAll != null && responses.size() != size) {
			String data = "require:" + size + " actual:" + responses.size();
			Cat.logEvent("FetchReportError:" + this.getClass().getSimpleName(), request.getDomain(), Event.SUCCESS, data);

			return null;
		}
		ModelResponse<T> aggregated = new ModelResponse<T>();
		T report = merge(request, responses);

		aggregated.setModel(report);
		return aggregated;
	}

	@Override
	public boolean isEligable(ModelRequest request) {
		for (ModelService<T> service : m_allServices) {
			if (service.isEligable(request)) {
				return true;
			}
		}

		return false;
	}

	protected abstract T merge(ModelRequest request, final List<ModelResponse<T>> responses);

	public void setSerivces(ModelService<T>... services) {
		for (ModelService<T> service : services) {
			m_allServices.add(service);
		}
	}

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder(64);

		sb.append(getClass().getSimpleName()).append('[');
		sb.append("name=").append(m_name);
		sb.append(']');

		return sb.toString();
	}
}

@@ -2354,35 +2354,23 @@
 int 
-size = m_allServices.size()
+requireSize = 0
 ;%0A%09%09
@@ -2471,20 +2471,16 @@
 nse%3CT%3E%3E(
-size
 ));%0A%09%09fi
@@ -2866,24 +2866,27 @@
 tinue;%0A%09%09%09%7D%0A
+%09%09%09
 %0A%09%09%09// save 
@@ -3069,16 +3069,37 @@
 );%0A%09%09%09%7D%0A
+%09%09%09requireSize++;%0A%09%09%09
 %0A%09%09%09s_th
@@ -3894,17 +3894,24 @@
 ze() != 
-s
+requireS
 ize) %7B%0A%09
@@ -3939,17 +3939,24 @@
 ire:%22 + 
-s
+requireS
 ize + %22 

fix bug