package com.ctrip.hermes.remoting;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

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

public class CommandProcessorManager implements Initializable {

	@Inject
	private CommandRegistry m_registry;

	private BlockingQueue<CommandContext> m_toProcess = new LinkedBlockingQueue<CommandContext>();

	private void process(CommandContext ctx) {
		Command cmd = ctx.getCommand();
		CommandProcessor processor = m_registry.findProcessor(cmd.getType());

		if (processor == null) {
			System.out.println(String.format("Command processor for type %s is not found", cmd.getType()));
		} else {
			try {
				processor.process(ctx);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	public void offer(CommandContext ctx) {
		m_toProcess.offer(ctx);
	}

	@Override
	public void initialize() throws InitializationException {
		// TODO
		new Thread() {

			@Override
			public void run() {
				while (true) {
					try {
						process(m_toProcess.take());
					} catch (InterruptedException e) {
						return;
					}
				}
			}

		}.start();
	}
}

@@ -100,35 +100,166 @@
 current.
-LinkedBlockingQueue
+ExecutorService;%0Aimport java.util.concurrent.LinkedBlockingQueue;%0Aimport java.util.concurrent.ThreadPoolExecutor;%0Aimport java.util.concurrent.TimeUnit
 ;%0A%0Aimpor
@@ -587,93 +587,34 @@
 ate 
-BlockingQueue%3CCommandContext%3E m_toProcess = new LinkedBlockingQueue%3CCommandContext%3E()
+ExecutorService m_executor
 ;%0A%0A%09
@@ -1021,16 +1021,22 @@
 d offer(
+final 
 CommandC
@@ -1057,316 +1057,358 @@
 %09%09m_
-toProcess.offer(ctx);%0A%09%7D%0A%0A%09@Override%0A%09public void initialize() throws InitializationException %7B%0A%09%09// TODO%0A%09%09new Thread() %7B%0A%0A%09%09%09@Override%0A%09%09%09public void run() %7B%0A%09%09%09%09while (true) %7B%0A%09%09%09%09%09try %7B%0A%09%09%09%09%09%09process(m_toProcess.take
+executor.submit(new Runnable() %7B%0A%0A%09%09%09@Override%0A%09%09%09public void run() %7B%0A%09%09%09%09process(ctx);%0A%09%09%09%7D%0A%09%09%7D);%0A%09%7D%0A%0A%09@Override%0A%09public void initialize() throws InitializationException %7B%0A%09%09// TODO%0A%09%09BlockingQueue%3CRunnable%3E workQueue = new LinkedBlockingQueue%3CRunnable%3E
 ()
-)
 ;%0A%09%09
-%09%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09%09%09%09return;%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%0A%09%09%7D.start(
+m_executor = new ThreadPoolExecutor(10, 10, Integer.MAX_VALUE, TimeUnit.SECONDS, workQueue
 );%0A%09

use thread pool to execute command