Package org.a2aproject.sdk.server.events

Interface QueueManager

All Known Implementing Classes:
InMemoryQueueManager, ReplicatedQueueManager
public interface QueueManager
Manages EventQueue lifecycle for task-based event routing and consumption.

The QueueManager is responsible for creating, storing, and managing event queues that coordinate asynchronous communication between agent executors (producers) and transport consumers. It supports both simple in-memory queuing and sophisticated patterns like queue tapping for resubscription and distributed event replication.

Queue Architecture

  • MainQueue: Primary queue for a task, created by createOrTap(String)
  • ChildQueue: Subscriber view created by tap(String), receives copies of parent events
  • One MainQueue per task, multiple ChildQueues for concurrent consumers (resubscription, cancellation)
  • Events enqueued to MainQueue are automatically distributed to all active ChildQueues

Queue Lifecycle

  1. Creation: createOrTap(String) creates MainQueue for new task or taps existing for resubscription
  2. Population: Agent enqueues events to MainQueue via EventQueue.enqueueEvent(org.a2aproject.sdk.spec.Event)
  3. Distribution: Events automatically copied to all ChildQueues
  4. Consumption: Consumers poll events from their queues (Main or Child)
  5. Closure: Queue closes on final event (COMPLETED/FAILED/CANCELED) or explicit close
  6. Cleanup: MainQueue removed from manager when all ChildQueues close and task is finalized

Default Implementation

InMemoryQueueManager provides the standard implementation:
  • Stores queues in thread-safe ConcurrentHashMap
  • Integrates with TaskStateProvider for cleanup decisions
  • Removes queues when tasks enter final state (COMPLETED/FAILED/CANCELED)
  • Supports queue tapping for resubscription scenarios

Alternative Implementations

  • extras/queue-manager-replicated: Kafka-based replication for multi-instance deployments
Replicated implementations enable event distribution across server instances for high availability and load balancing.

Tapping Pattern (Resubscription)

Tapping creates a ChildQueue that receives future events from an ongoing task: 

 // Client disconnects and later reconnects
 EventQueue childQueue = queueManager.tap(taskId);
 if (childQueue != null) {
     // Receive events from this point forward
     // (Historical events before tap are not replayed)
 }
Use cases:
  • Resubscribing to ongoing tasks after disconnect
  • Canceling tasks while still receiving status updates
  • Multiple concurrent consumers of the same task

CDI Extension Pattern


 @ApplicationScoped
 @Alternative
 @Priority(50)  // Higher than default InMemoryQueueManager
 public class KafkaQueueManager implements QueueManager {
     // Custom implementation with event replication
 }

Thread Safety

All methods must be thread-safe. Multiple threads may call createOrTap(), tap(), and close() concurrently for different tasks.
See Also:

  • Method Details

    • add

      void add(String taskId, EventQueue queue)
      Adds a queue to the manager with the given task ID.

      Throws TaskQueueExistsException if a queue already exists for this task. Typically used internally - prefer createOrTap(String) for most use cases.

      Parameters:
      taskId - the task identifier
      queue - the queue to add
      Throws:
      TaskQueueExistsException - if queue already exists for this task ID

    • get

      @Nullable EventQueue get(String taskId)
      Retrieves the MainQueue for a task, if it exists.

      Returns the primary queue for the task. Does not create a new queue if none exists.

      Parameters:
      taskId - the task identifier
      Returns:
      the MainQueue, or null if no queue exists for this task

    • tap

      @Nullable EventQueue tap(String taskId)
      Creates a ChildQueue that receives copies of events from the MainQueue.

      Use this for:

      • Resubscribing to an ongoing task (receive future events)
      • Canceling a task while still receiving status updates
      • Multiple concurrent consumers of the same task

      The ChildQueue receives events enqueued AFTER it's created. Historical events are not replayed.

      Parameters:
      taskId - the task identifier
      Returns:
      a ChildQueue that receives future events, or null if the MainQueue doesn't exist

    • close

      void close(String taskId)
      Closes and removes the queue for a task.

      This closes the MainQueue and all ChildQueues, then removes it from the manager. Called during cleanup after task completion or error conditions.

      Parameters:
      taskId - the task identifier

    • createOrTap

      EventQueue createOrTap(String taskId)
      Creates a MainQueue if none exists, or taps the existing queue to create a ChildQueue.

      This is the primary method used by DefaultRequestHandler:

      • New task: Creates and returns a MainQueue
      • Resubscription: Taps existing MainQueue and returns a ChildQueue
      Parameters:
      taskId - the task identifier
      Returns:
      a MainQueue (if new task) or ChildQueue (if tapping existing)

    • awaitQueuePollerStart

      void awaitQueuePollerStart(EventQueue eventQueue) throws InterruptedException
      Waits for the queue's consumer polling to start.

      Used internally to ensure the consumer is ready before the agent starts enqueueing events, avoiding race conditions where events might be enqueued before the consumer begins polling.

      Parameters:
      eventQueue - the queue to wait for
      Throws:
      InterruptedException - if interrupted while waiting

    • getEventQueueBuilder

      default EventQueue.EventQueueBuilder getEventQueueBuilder(String taskId)
      Returns an EventQueueBuilder for creating queues with task-specific configuration.

      Implementations can override to provide custom queue configurations per task, such as different capacities, hooks, or event processors.

      Default implementation returns a standard builder with no customization.

      Parameters:
      taskId - the task ID for context (may be used to customize queue configuration)
      Returns:
      a builder for creating event queues

    • createBaseEventQueueBuilder

      default EventQueue.EventQueueBuilder createBaseEventQueueBuilder(String taskId)
      Creates a base EventQueueBuilder with standard configuration for this QueueManager. This method provides the foundation for creating event queues with proper configuration (MainEventBus, TaskStateProvider, cleanup callbacks, etc.).

      QueueManager implementations that use custom factories can call this method directly to get the base builder without going through the factory (which could cause infinite recursion if the factory delegates back to getEventQueueBuilder()).

      Callers can then add additional configuration (hooks, callbacks) before building the queue.

      Parameters:
      taskId - the task ID for the queue
      Returns:
      a builder with base configuration specific to this QueueManager implementation

    • getActiveChildQueueCount

      int getActiveChildQueueCount(String taskId)
      Returns the number of active ChildQueues for a task.

      Used for testing to verify reference counting and queue lifecycle management. In production, indicates how many consumers are actively subscribed to a task's events.

      Parameters:
      taskId - the task ID
      Returns:
      number of active child queues, or -1 if the MainQueue doesn't exist