package org.craft.atom.rpc;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import lombok.Getter;
import lombok.Setter;

import org.craft.atom.protocol.rpc.model.RpcMessage;
import org.craft.atom.rpc.spi.RpcExecutorFactory;
import org.craft.atom.rpc.spi.RpcInvoker;
import org.craft.atom.rpc.spi.RpcProcessor;

/**
 * @author mindwind
 * @version 1.0, Aug 7, 2014
 */
public class DefaultRpcProcessor implements RpcProcessor {
		
	
	@Getter @Setter private RpcInvoker         invoker        ;
	@Getter @Setter private RpcExecutorFactory executorFactory;
	
	
	@Override
	public RpcMessage process(final RpcMessage req) {
		ExecutorService executor = executor(req);
		
		try {
			Future<RpcMessage> future = executor.submit(new Callable<RpcMessage>() {
				@Override
				public RpcMessage call() throws Exception {
					return process0(req);
				}
			});
			return future.get(rpcTimeoutInMillis(req), TimeUnit.MILLISECONDS);
		} catch (ExecutionException e) {
			return RpcMessages.newRsponseRpcMessage(new RpcException(RpcException.BUSINESS, e));
		} catch (TimeoutException e) {
			return RpcMessages.newRsponseRpcMessage(new RpcException(RpcException.TIMEOUT, e));
		} catch (RejectedExecutionException e) {
			return RpcMessages.newRsponseRpcMessage(new RpcException(RpcException.OVERLOAD, e));
		} catch (Exception e) {
			return RpcMessages.newRsponseRpcMessage(new RpcException(RpcException.UNKNOWN, e));
		}
	}
	
	private RpcMessage process0(RpcMessage req) {
		RpcMessage rsp;
		try {
			rsp = invoker.invoke(req);
		} catch (RpcException e) {
			rsp = RpcMessages.newRsponseRpcMessage(e);
		}
		return rsp;
	}
	
	private long rpcTimeoutInMillis(RpcMessage req) {
		long timeout = req.getBody().getRpcOption().getRpcTimeoutInMillis();
		if (timeout == 0) { timeout = Long.MAX_VALUE; }
		return timeout;
	}
	
	private ExecutorService executor(RpcMessage rm) {
		return executorFactory.getExecutor(rm);
	}

}

@@ -1289,16 +1289,20 @@
 ion.
-BUSINESS
+SERVER_ERROR
 , e)
@@ -1410,16 +1410,23 @@
 ception.
+SERVER_
 TIMEOUT,
@@ -1547,16 +1547,23 @@
 ception.
+SERVER_
 OVERLOAD
@@ -1565,24 +1565,100 @@
 RLOAD, e));%0A
+%09%09%7D catch (RpcException e) %7B%0A%09%09%09return RpcMessages.newRsponseRpcMessage(e);%0A
 %09%09%7D catch (E

improve - RpcException handle