Class Agent

src/jvm/clojure/lang/Agent.java:22–287

import java.util.concurrent.atomic.AtomicReference;

public class Agent extends ARef {

static class ActionQueue {
	public final IPersistentStack q;
	public final Throwable error; // non-null indicates fail state
	static final ActionQueue EMPTY = new ActionQueue(PersistentQueue.EMPTY, null);

	public ActionQueue( IPersistentStack q, Throwable error )
		{
		this.q = q;
		this.error = error;
		}
}

static final Keyword CONTINUE = Keyword.intern(null, "continue");
static final Keyword FAIL = Keyword.intern(null, "fail");

volatile Object state;
    AtomicReference<ActionQueue> aq = new AtomicReference<ActionQueue>(ActionQueue.EMPTY);

    volatile Keyword errorMode = CONTINUE;
    volatile IFn errorHandler = null;

final private static AtomicLong sendThreadPoolCounter = new AtomicLong(0);

final private static AtomicLong sendOffThreadPoolCounter = new AtomicLong(0);

volatile public static ExecutorService pooledExecutor =
	Executors.newFixedThreadPool(2 + Runtime.getRuntime().availableProcessors(), 
		createThreadFactory("clojure-agent-send-pool-%d", sendThreadPoolCounter));

volatile public static ExecutorService soloExecutor = Executors.newCachedThreadPool(
	createThreadFactory("clojure-agent-send-off-pool-%d", sendOffThreadPoolCounter));

final static ThreadLocal<IPersistentVector> nested = new ThreadLocal<IPersistentVector>();

private static ThreadFactory createThreadFactory(final String format, final AtomicLong threadPoolCounter) {
	return new ThreadFactory() {
		public Thread newThread(Runnable runnable) {
			Thread thread = new Thread(runnable);
			thread.setName(String.format(format, threadPoolCounter.getAndIncrement()));
			return thread;
		}
	};
}

public static void shutdown(){
	soloExecutor.shutdown();
	pooledExecutor.shutdown();
}

static class Action implements Runnable{
	final Agent agent;
	final IFn fn;
	final ISeq args;
	final Executor exec;