Package org.a2aproject.sdk.server.events

Class InMemoryQueueManager

java.lang.Object
org.a2aproject.sdk.server.events.InMemoryQueueManager
All Implemented Interfaces:
QueueManager
@ApplicationScoped public class InMemoryQueueManager extends Object implements QueueManager

  • Constructor Details

    • InMemoryQueueManager

      protected InMemoryQueueManager()
      No-args constructor for CDI proxy creation. CDI requires a non-private constructor to create proxies for @ApplicationScoped beans. All fields are initialized by the @Inject constructor during actual bean creation.

    • InMemoryQueueManager

      @Inject public InMemoryQueueManager(TaskStateProvider taskStateProvider, MainEventBus mainEventBus)

    • InMemoryQueueManager

      public InMemoryQueueManager(EventQueueFactory factory, TaskStateProvider taskStateProvider, MainEventBus mainEventBus)

  • Method Details

    • add

      public void add(String taskId, EventQueue queue)
      Description copied from interface: QueueManager
      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 QueueManager.createOrTap(String) for most use cases.

      Specified by:
      add in interface QueueManager
      Parameters:
      taskId - the task identifier
      queue - the queue to add

    • get

      public @Nullable EventQueue get(String taskId)
      Description copied from interface: QueueManager
      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.

      Specified by:
      get in interface QueueManager
      Parameters:
      taskId - the task identifier
      Returns:
      the MainQueue, or null if no queue exists for this task

    • tap

      public @Nullable EventQueue tap(String taskId)
      Description copied from interface: QueueManager
      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.

      Specified by:
      tap in interface QueueManager
      Parameters:
      taskId - the task identifier
      Returns:
      a ChildQueue that receives future events, or null if the MainQueue doesn't exist

    • close

      public void close(String taskId)
      Description copied from interface: QueueManager
      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.

      Specified by:
      close in interface QueueManager
      Parameters:
      taskId - the task identifier

    • createOrTap

      public EventQueue createOrTap(String taskId)
      Description copied from interface: QueueManager
      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
      Specified by:
      createOrTap in interface QueueManager
      Parameters:
      taskId - the task identifier
      Returns:
      a MainQueue (if new task) or ChildQueue (if tapping existing)

    • awaitQueuePollerStart

      public void awaitQueuePollerStart(EventQueue eventQueue) throws InterruptedException
      Description copied from interface: QueueManager
      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.

      Specified by:
      awaitQueuePollerStart in interface QueueManager
      Parameters:
      eventQueue - the queue to wait for
      Throws:
      InterruptedException - if interrupted while waiting

    • getEventQueueBuilder

      public EventQueue.EventQueueBuilder getEventQueueBuilder(String taskId)
      Description copied from interface: QueueManager
      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.

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

    • getActiveChildQueueCount

      public int getActiveChildQueueCount(String taskId)
      Description copied from interface: QueueManager
      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.

      Specified by:
      getActiveChildQueueCount in interface QueueManager
      Parameters:
      taskId - the task ID
      Returns:
      number of active child queues, or -1 if the MainQueue doesn't exist

    • createBaseEventQueueBuilder

      public EventQueue.EventQueueBuilder createBaseEventQueueBuilder(String taskId)
      Description copied from interface: QueueManager
      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.

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

    • getCleanupCallback

      public Runnable getCleanupCallback(String taskId)
      Get the cleanup callback that removes a queue from the map when it closes. This is exposed so that subclasses (like ReplicatedQueueManager) can reuse this cleanup logic while adding their own callbacks in the correct order.

      The cleanup callback checks if the task is finalized before removing the queue. If the task is not finalized, the queue remains in the map to handle late-arriving events.

      Parameters:
      taskId - the task ID for the queue
      Returns:
      a Runnable that removes the queue from the map if appropriate