EDF

package czp.opensource.edf;

public class EDFEvent {

	/**
	 * 事件源
	 */
	private Object from;
	/**
	 * 异常
	 */
	private Throwable error;

	/**
	 * 事件的具体内容
	 */
	private Object body;
	
	/**
	 * 事件类型
	 */
	private EDFEventType type;
	
	
	public EDFEvent(Object from, Throwable error, Object body, EDFEventType type) {
		this.from = from;
		this.error = error;
		this.body = body;
		this.type = type;
	}
	
	public EDFEvent(Object body, EDFEventType type) {
		this.body = body;
		this.type = type;
	}

	public EDFEvent() {
	}

	/**
	 * 
	 * @return 事件的内容
	 */
	public Object getBody() {
		return body;
	}

	/**
	 * 
	 * @param body 事件的内容
	 */
	public void setBody(Object body) {
		this.body = body;
	}

	/**
	 * 
	 * @param from 事件的发送者
	 */
	public void setFrom(Object from) {
        this.from = from;		
	}

	/**
	 * 
	 * @param error 事件的异常
	 */
	public void setError(Throwable error) {
		this.error = error;
	}


	/**
	 * 
	 * @return 事件的发送者
	 */
	public Object getFrom() {
		return from;
	}

	/**
	 * 
	 * @return 事件异常
	 */
	public Throwable getError() {
		return error;
	}

	/**
	 * 
	 * @return 事件类型
	 */
	public EDFEventType getType() {
		return this.type;
	}

	/**
	 * 
	 * @param type 事件类型
	 */
	public void setType(EDFEventType type) {
		this.type = type;
	}

	@Override
	public String toString() {
		StringBuilder str = new StringBuilder();
		str.append("EDFEvent from:").append(from).append(",error:").append(error)
		.append(",type:").append(type).append(",body:").append(body);
		return str.toString();
	}
   
}
//---------

package czp.opensource.edf;

/**
 * 事件的处理器
 * 
 * @author CaoZhongping
 *
 */
public interface EDFEventHandler {

	/**
	 * 处理具体的事件
	 * @param event  容器分配的事件
	 * @return Object 返回值 
	 */
	Object handle(EDFEvent event);


	/**
	 * 处理异常事件
	 * @param event 容器分配的异常事件
	 */
	void handdleError(EDFEvent event);

	/**
	 * @return EDFEventType 关注的事件类型
	 */
	EDFEventType getMatchType();
}
//------

package czp.opensource.edf;

/**
 * 事件类型接口默认提供两个实现:
 * <pre>
 * 1 错误类型ERROR_TYPE,当处理事件发生
 *   异常时会发送该类型的事件并调用handle的handleError
 *   调用者一般不需要发送异常类型的事件
 * 2 所有类型ALL_TYPE,如果一个handle注册了ERROR_TYPE
 *   则表示该关注所有事件,因为isMatch恒返回true
 * 使用者需要实现  getType()和isMatch(),例如:
 * 
 *   public class SimpleEDFEventType implements EDFEventType {

		private String myType;
		
		public SimpleEDFEventType(String myType) {
			this.myType = myType;
		}
	
		@Override
		public String getType() {
			return myType;
		}
	
		@Override
		public boolean isMatch(EDFEventType type) {
			return type.getType().equalsIgnoreCase(getType());
		}

	}
 * 
 * </pre>
 * 
 * @author CaoZhongping
 *
 */
public interface EDFEventType {

	/**
	 * @return 事件类型的名称
	 */
	String getType();
	
	/**
	 * 反应器在处理事件时会调用
	 * 该方法找到该事件的handler
	 * 这个方法应该尽可能的简单
	 * 
	 * @param type 事件类型
	 * @return boolean 是否匹配
	 */
	boolean isMatch(EDFEventType type);
	
	/**
	 * 错误事件类型
	 */
	EDFEventType ERROR_TYPE = new EDFEventType(){

		private String type="EDFEventType_ERROR";
		
		@Override
		public String getType() {
			return type;
		}

		@Override
		public boolean isMatch(EDFEventType eventType) {
			return eventType.getType().equals(getType());
		}
		
	};
	/**
	 * 所有的事件类型isMatch恒为true
	 * 也就是关注所有的事件
	 */
	EDFEventType ALL_TYPE = new EDFEventType() {
		private String type="EDFEventType_ALL";
		@Override
		public boolean isMatch(EDFEventType type) {
			return true;
		}
		
		@Override
		public String getType() {
			return type;
		}
	};
}
//-----------

package czp.opensource.edf;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


/**
 * 事件驱动框架的事件反应器
 * 
 * @author CaoZhongping
 *
 */
public class EDFReactor implements Runnable{

	/**
	 * 每个处理器默认的处理超时时间
	 */
	private static final long DEFAULT_TIME_OUT_SEC = 15;

	/**
	 * 默认的队列大小
	 */
	private final static int DEFAULT_QUEUE_CAPCITY = 50;
	/**
	 * 事件队列
	 */
	private BlockingQueue<EDFEvent> queue;
	
	/**
	 * 分派线程池
	 */
	private ExecutorService dispatchService;
	
	/**
	 * 工作线程池
	 */
	private ExecutorService workService;

	/**
	 * 队列容量
	 */
	private int eventQueueCapability;

	/**
	 * 事件处理器
	 */
	private Map<String, List<EDFEventHandler>> handlers;

	/**
	 * 创建反应器的线程
	 */
	private Thread createThread;
	
	private volatile boolean isStop;
	
	public EDFReactor(ExecutorService dispatchService,
			ExecutorService workService,int  eventQueueMaxSize) {
		this.dispatchService = dispatchService;
		this.workService = workService;
		this.queue = new ArrayBlockingQueue<EDFEvent>(eventQueueMaxSize);
		this.handlers = new ConcurrentHashMap<String, List<EDFEventHandler>>();
		this.createThread = Thread.currentThread();
	}

	public EDFReactor()
	{
		this(Executors.newSingleThreadExecutor(), Executors.newCachedThreadPool(), DEFAULT_QUEUE_CAPCITY);
	}
	/**
	 * 添加处理器
	 * @param eventHandler
	 */
	public void addEventHandler(EDFEventHandler eventHandler)
	{
		String type = eventHandler.getMatchType().getType();
		List<EDFEventHandler> ls = handlers.get(type);
		if(ls==null)
		{
			ls = new CopyOnWriteArrayList<EDFEventHandler>();
			ls.add(eventHandler);
			handlers.put(type, ls);
		}else{
			ls.add(eventHandler);
		}
	}
	/**
	 * 启动反应器
	 */
	public void start(){
		if(dispatchService.isShutdown())
		{
			throw new RuntimeException("the EDFReactor has Shutdown ,Please create a new EDFReactor");
		}
		this.dispatchService.execute(this);
		
	}

	/**
	 * 停止反应器
	 * 只有反应器的创建线程可以终止它
	 */
	public void stop()
	{
		if(Thread.currentThread().equals(createThread))
		{
			this.isStop = true;
			this.queue.clear();
			this.handlers.clear();
			this.dispatchService.shutdownNow();
			this.workService.shutdownNow();
		}else{
			throw new RuntimeException("EDFReactor stop() only be called by create thread");
		}
	}
	

    /**
     * @return 队列的容量
     */
	public int getEventQueueCapability() {
		return eventQueueCapability;
	}

	/**
	 *添加事件到队列
	 * @param event
	 */
	public void sendEvent(EDFEvent event)
	{
		this.queue.offer(event);
	}
	/**
	 * 分派线程开始
	 */
	@Override
	public void run() {
		while(!isStop)
		{
			try {
				EDFEvent ev = queue.take();
				handleEvent(ev);
			} catch (Exception e) {
				sendExceptionEvent(e);
			}
			
		}
	}

	/**
	 * 在工作线程里调用具体的事件处理器
	 * 
	 * @param ev
	 */
	private void handleEvent(EDFEvent ev) {
		List<EDFEventHandler> ls = handlers.get(ev.getType().getType());
		if(ls!=null)
		{
			for(EDFEventHandler hdl:ls)
			{
				Future<?> future = null;
				try {
					final EDFEventTask task = new EDFEventTask(ev,hdl);
					future = workService.submit(task);
					if(!future.isDone())
					{
						future.get(DEFAULT_TIME_OUT_SEC, TimeUnit.SECONDS);
					}
					
				}catch (Exception e) {
					sendExceptionEvent(e);
				}
			}
		}
	}

	/**
	 * 发送异常事件
	 * @param throwable
	 */
	public void sendExceptionEvent(Throwable throwable) {
		EDFEvent event = new EDFEvent();
		event.setFrom(this);
		event.setError(throwable);
		event.setType(EDFEventType.ERROR_TYPE);
		sendEvent(event);
	}
	/**
	 * 具体的事件
	 * 
	 * @author CaoZhongping
	 *
	 */
	private static  class EDFEventTask implements Callable<Object> {

		private EDFEvent event;
		private EDFEventHandler handle;

		public EDFEventTask(EDFEvent event, EDFEventHandler handle) {
			this.event = event;
			this.handle = handle;
			
		}

		@Override
		public Object call() throws Exception {
			if(event.getType().isMatch(EDFEventType.ERROR_TYPE))
			{
				handle.handdleError(event);
			}else if(handle.getMatchType().isMatch(event.getType()))
			{
				return handle.handle(event);
			}
			return null;
		}
	}
}
//---------

package czp.opensource.edf.demo;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import czp.opensource.edf.EDFEvent;
import czp.opensource.edf.EDFEventHandler;
import czp.opensource.edf.EDFEventType;
import czp.opensource.edf.EDFReactor;

/**
 * <pre>
 *   参见:czp.opensource.edf.demo.SimpleDemo
 * </pre>
 * @author CaoZhongping
 *
 */
public class SimpleDemo {

	public static void main(String[] args) throws InterruptedException {
		
		//创建一个反应器通常一个系统创建一个reactor
		final EDFReactor reactor = new EDFReactor();
		//添加一个处理器
		reactor.addEventHandler(new SimpleEDFEventHandle());
		//启动反应器
		reactor.start();
		
		final CountDownLatch sd = new CountDownLatch(1);
		ExecutorService service = Executors.newFixedThreadPool(5);
		
		service.execute(new Runnable() {
			
			@Override
			public void run() {
				int i = 0;
				while(i++<20)
				{
					try {
						//创建事件
						EDFEvent event = new EDFEvent();
						event.setType(new SimpleEDFEventType("test"));
						event.setFrom(this);
						//发送事件
						reactor.sendEvent(event );
						Thread.sleep(10);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
				sd.countDown();
				
			}
		});
		sd.await();
		service.shutdownNow();
		//系统注销时stop反应器
		reactor.stop();
	}
}
/**
 * 简单的处理器
 * 
 * @author CaoZhongping
 *
 */
class SimpleEDFEventHandle implements EDFEventHandler {

	private EDFEventType myType = new SimpleEDFEventType("test");
	
	@Override
	public Object handle(EDFEvent event) {
		System.out.println("get msg "+event);
		return null;
	}

	@Override
	public void handdleError(EDFEvent event) {
     System.out.println(event.getError());
	}

	@Override
	public EDFEventType getMatchType() {
		return myType;
	}
    
}
/**
 * 简单的事件类型
 * 
 * @author CaoZhongping
 */
class SimpleEDFEventType implements EDFEventType {

	private String myType;
	
	public SimpleEDFEventType(String myType) {
		this.myType = myType;
	}

	@Override
	public String getType() {
		return myType;
	}

	@Override
	public boolean isMatch(EDFEventType type) {
		return type.getType().equalsIgnoreCase(getType());
	}

	@Override
	public String toString() {
		return "SimpleEDFEventType [myType=" + myType + "]";
	}
	
}