stagemanager.java

sea是一个基于seda模式的实现。这个设计模式将系统分为很多stage。每个stage分布不同的任务（基于线程池）。通过任务流的方式提高系统的效率。

/* 
Copyright (c) 2003, The Regents of the University of California, through 
Lawrence Berkeley National Laboratory (subject to receipt of any required 
approvals from the U.S. Dept. of Energy).  All rights reserved.
*/
package gov.lbl.dsd.sea;

import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;

import EDU.oswego.cs.dl.util.concurrent.DirectExecutor;
import EDU.oswego.cs.dl.util.concurrent.Executor;
import EDU.oswego.cs.dl.util.concurrent.LinkedQueue;
import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
import EDU.oswego.cs.dl.util.concurrent.QueuedExecutor;
import EDU.oswego.cs.dl.util.concurrent.ThreadedExecutor;


/**
 * Creates and manages a set of stages;
 * Contains several default factories for commonly used threading
 * policies. By far the most commonly used are <code>QUEUED</code>,
 * <code>POOLED</code> and <code>DIRECT</code>.
 * 
 * @author [e-mail]
 * @author $Author: gegles $
 * @version $Revision: 1.17 $, $Date: 2004/09/16 16:57:15 $
 */
public class StageManager {

	//private static StageManager queuedManager; 
	protected final Map stages; // the stages created by this class
	protected final ExecutorFactory executorFactory;

	/** 
	 * Creates an instance with the default threading policy. 
	 */
	public StageManager() {
		this(QUEUED);
	}
	
	/** 
	 * Creates an instance with the given threading policy. 
	 */
	public StageManager(ExecutorFactory executorFactory) {
		if (executorFactory == null) throw new IllegalArgumentException("executor factory must not be null");
		this.executorFactory = executorFactory;
		this.stages = new LinkedHashMap();
	}
	
	/** 
	 * Adds the given stage to the set of managed stages.
	 * @return the same stage (for convenience)
	 */
	public synchronized Stage addStage(Stage stage) {
		String stageName = stage.getName();
		if (this.getStage(stageName) != null) {
			throw new IllegalArgumentException("stage=" + stageName + " already added previously.");
		}
		
		this.stages.put(stageName, stage);
		return stage;
	}
	
	/** 
	 * Creates and returns a managed stage with the given event handler. 
	 */
	public synchronized Stage createStage(EventHandler handler) {
		return this.createStage(handler.getClass().getName() + "#" + this.stages.size(), handler);
	}

	/** 
	 * Creates and returns a managed stage with the given name and event handler. 
	 */
	public Stage createStage(String stageName, EventHandler handler) {
		return this.createStage(stageName, handler, createExceptionHandler());
	}

	/** 
	 * Creates and returns a managed stage with the given name, event handler and exceptionHandler. 
	 */
	public Stage createStage(String stageName, EventHandler handler, ExceptionHandler exceptionHandler) {
		return this.createStage(stageName, handler, exceptionHandler, this.executorFactory);
	}

	/** 
	 * Creates and returns a managed stage with the given name, event handler, exceptionHandler and executor. 
	 */
	protected Stage createStage(String stageName, EventHandler handler, ExceptionHandler exceptionHandler, ExecutorFactory executorFactory) {
		Stage stage = new Stage(stageName, handler, executorFactory, exceptionHandler);
		this.addStage(stage);
		return stage;
	}
	
	/** 
	 * Override this method for custom exception handlers. 
	 */
	protected ExceptionHandler createExceptionHandler() {
		return null;
	}
	
	/** 
	 * Starts all managed stages.
	 */
	public synchronized void startAll() {
		Iterator iter = this.stages.values().iterator();
		while (iter.hasNext()) {
			Stage stage = (Stage) iter.next();
			stage.start();
		}
	}

	/** 
	 * Stops all managed stages.
	 */
	public synchronized void stopAll() {
		Iterator iter = this.stages.values().iterator();
		while (iter.hasNext()) {
			Stage stage = (Stage) iter.next();
			stage.stop();
		}
	}
	
//	/**
//	 * Returns the global queued default instance of this class. 
//	 */
//	public static synchronized StageManager getManager() {
//		if (queuedManager == null) queuedManager = new StageManager(QUEUED);
//		return queuedManager;
//	}
	
	/** 
	 * Returns the managed stage with the given name, 
	 * or <code>null</code> if no such stage is currently managed.
	 */
	public synchronized Stage getStage(String stageName) {
		return (Stage) this.stages.get(stageName);
	}
	
	/**
	 * Returns a string representation of the receiver.
	 */
	public synchronized String toString() {
		return super.toString() + ": execFactory=" + this.executorFactory +
				", stages=" + this.stages.values().toString();
	}


////////////////////////////////////////////////////////////////
// nested classes
////////////////////////////////////////////////////////////////

	/**
	 * Creates and returns {@link DirectExecutor} instances.
	 */
	public static final ExecutorFactory DIRECT = new Direct();

	
	private static class Direct implements ExecutorFactory {		
		public Executor createExecutor() {
			return new DirectExecutor();
		}
	}

	/**
	 * Creates and returns {@link QueuedExecutor} instances.
	 */
	public static final ExecutorFactory QUEUED = new Queued();

	private static class Queued implements ExecutorFactory {		
		public Executor createExecutor() {
			return new QueuedExecutor(new LinkedQueue());
//			Channel ch; 
//			ch = new LinkedQueue();
//			//ch = new BoundedLinkedQueue(100);
//			//ch = new BoundedPriorityQueue(100);
//			//ch = new Slot();
//			
//			Executor exec;
//			exec = new QueuedExecutor(ch); 
//			//exec = new DirectExecutor();
//			//exec = new ThreadedExecutor();
//			//exec = new PooledExecutor(ch);
//			//exec = new LockedExecutor(new Mutex());
//			return exec;
		}
	}

	/**
	 * Creates and returns {@link ThreadedExecutor} instances.
	 */
	public static final ExecutorFactory THREADED = new Threaded();

	private static class Threaded implements ExecutorFactory {		
		public Executor createExecutor() {
			return new ThreadedExecutor();
		}
	}

	/**
	 * Creates and returns {@link PooledExecutor} instances.
	 */
	public static final ExecutorFactory POOLED = new Pooled();

	private static class Pooled implements ExecutorFactory {		
		public Executor createExecutor() {
			PooledExecutor pexec = new PooledExecutor(new LinkedQueue());
			pexec.setKeepAliveTime(1000);
			return pexec;
		}
	}

}