Class Agent

src/main/java/groovy/concurrent/Agent.java:77–508

A thread-safe mutable-value container inspired by Clojure's agents and GPars' Agent. An Agent wraps a value that can be read by any thread but modified only through serialized update functions. Updates are queued and applied one at a time on a dedicated executor, guaranteeing tha

 * @since 6.0.0
 */
public final class Agent<T> {

    /** Default per-subscriber buffer size for {@link #changes()}. */
    private static final int DEFAULT_CHANGES_BUFFER = 256;

    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    private final ExecutorService updateExecutor;
    private final Object lifecycleLock = new Object();
    private volatile T value;
    private volatile SubmissionPublisher<T> changesPublisher;
    private volatile boolean shutdownInvoked;

    private Agent(T initialValue, ExecutorService executor) {
        this.value = initialValue;
        this.updateExecutor = executor;
    }

    /**
     * Creates an agent with the given initial value, using a
     * single-thread executor for serialized updates.
     *
     * @param initialValue the starting value
     * @param <T>          the value type
     * @return a new agent
     */
    public static <T> Agent<T> create(T initialValue) {
        return new Agent<>(initialValue,
                Executors.newSingleThreadExecutor(r -> {
                    Thread t = new Thread(r, "groovy-agent");
                    t.setDaemon(true);
                    return t;
                }));
    }

    /**
     * Creates an agent backed by the given pool for update execution.
     * Updates are still serialized (only one at a time), but they run
     * on the pool's threads.
     *
     * @param initialValue the starting value
     * @param pool         the pool to use for updates
     * @param <T>          the value type
     * @return a new agent
     */
    public static <T> Agent<T> create(T initialValue, Pool pool) {
        Objects.requireNonNull(pool, "pool must not be null");
        // Use a SerialExecutor to serialize updates on the pool's threads.
        // We cannot use newSingleThreadExecutor with a delegating ThreadFactory
        // because that breaks the executor's internal task loop.
        return new Agent<>(initialValue, new SerialExecutor(pool));
    }

    /**
     * Returns the current value. This is a non-blocking snapshot read.
     *
     * @return the current value
     */
    public T get() {