package org.craft.atom.rpc;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.craft.atom.protocol.rpc.model.RpcMessage;
import org.craft.atom.protocol.rpc.model.RpcMethod;
import org.craft.atom.rpc.spi.RpcExecutorFactory;
import org.craft.atom.util.NamedThreadFactory;

/**
 * @author mindwind
 * @version 1.0, Aug 12, 2014
 */
public class DefaultRpcExecutorFactory implements RpcExecutorFactory {

	
	private Map<String, ExecutorService> pool = new ConcurrentHashMap<String, ExecutorService>();
	
	
	@Override
	public ExecutorService getExecutor(RpcMessage rm) {
		Class<?> rpcInterface = rm.getBody().getRpcInterface();
		RpcMethod rpcMethod = rm.getBody().getRpcMethod();
		String key = Integer.toString(rpcInterface.hashCode()) + Integer.toString(rpcMethod.hashCode());
		return getExecutor(key);
	}
	
	private ExecutorService getExecutor(String key) {
		ExecutorService es = pool.get(key);
		if (es == null) {
			synchronized (this) {
				if (es == null) {
					// TODO thread parameter config
					ThreadPoolExecutor tpe = new ThreadPoolExecutor(10, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("craft-atom-rpc"));
					tpe.allowCoreThreadTimeOut(true);
					es = tpe;
					pool.put(key, tpe);
				}
			}
		}
		return es;
	}

}

@@ -377,16 +377,68 @@
 Method;%0A
+import org.craft.atom.protocol.rpc.model.RpcOption;%0A
 import o
@@ -702,16 +702,20 @@
 e%3E pool 
+    
 = new Co
@@ -758,16 +758,115 @@
 ice%3E();%0A
+%09private RpcRegistry                  registry = RpcRegistry.getInstance()                       ;%0A
 %09%0A%09%0A%09@Ov
@@ -935,16 +935,17 @@
 Class%3C?%3E
+ 
  rpcInte
@@ -1056,89 +1056,44 @@
 y = 
-Integer.toString(rpcInterface.hashCode()) + Integer.toString(rpcMethod.hashCode()
+registry.key(rpcInterface, rpcMethod
 );%0A%09
@@ -1288,39 +1288,198 @@
 %09%09%09%09
-// TODO thread parameter config
+RpcEntry  entry   = registry.lookup(key)  ;%0A%09%09%09%09%09RpcOption option  = entry.getRpcOption()  ;%0A%09%09%09%09%09int       threads = option.getRpcThreads();%0A%09%09%09%09%09int       queues  = option.getRpcQueues() ;
 %0A%09%09%09
@@ -1532,14 +1532,24 @@
 tor(
-10, 10
+threads, threads
 , 60
@@ -1602,16 +1602,22 @@
 nnable%3E(
+queues
 ), new N

improve - use rpc option parameter for executor