Package org.a2aproject.sdk.server.events

Class EventQueue

java.lang.Object
org.a2aproject.sdk.server.events.EventQueue
All Implemented Interfaces:
AutoCloseable
public abstract class EventQueue extends Object implements AutoCloseable
Abstract base class for event queues that manage task event streaming.

An EventQueue provides a thread-safe mechanism for enqueueing and dequeueing events related to task execution. It supports backpressure through semaphore-based throttling and hierarchical queue structures via MainQueue and ChildQueue implementations.

Use builder(MainEventBus) to create configured instances or extend MainQueue/ChildQueue directly.

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    static class 
    EventQueue.EventQueueBuilder
    Builder for creating configured EventQueue instances.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    static final int
    DEFAULT_QUEUE_SIZE
    Default maximum queue size for event queues.

  • Constructor Summary

    Constructors
    Modifier
    Constructor
    Description
    protected
    EventQueue()
    Creates an EventQueue with the default queue size.
    protected
    EventQueue(int queueSize)
    Creates an EventQueue with the specified queue size.
    protected
    EventQueue(EventQueue parent)
    Creates an EventQueue as a child of the specified parent queue.

  • Method Summary

    Modifier and Type
    Method
    Description
    abstract void
    awaitQueuePollerStart()
    Waits for the queue poller to start consuming events.
    void
    clearAwaitingFinalEvent()
    Clears the awaiting final event flag.
    abstract void
    close()
    Closes this event queue gracefully, allowing pending events to be consumed.
    abstract void
    close(boolean immediate)
    Closes this event queue with control over immediate shutdown.
    abstract void
    close(boolean immediate, boolean notifyParent)
    Close this queue with control over parent notification (ChildQueue only).
    abstract @Nullable EventQueueItem
    dequeueEventItem(int waitMilliSeconds)
    Dequeues an EventQueueItem from the queue.
    protected void
    doClose()
    Internal method to close the queue gracefully.
    protected void
    doClose(boolean immediate)
    Internal method to close the queue with control over immediate shutdown.
    void
    enqueueEvent(Event event)
    Enqueues an event for processing.
    void
    enqueueEventLocalOnly(Event event)
    Enqueues an event directly to this specific queue only, bypassing the MainEventBus.
    abstract void
    enqueueItem(EventQueueItem item)
    Enqueues an event queue item for processing.
    void
    enqueueLocalOnly(EventQueueItem item)
    Enqueues an event directly to this specific queue only, bypassing the MainEventBus.
    int
    getQueueSize()
    Returns the configured queue size.
    boolean
    isAwaitingFinalEvent()
    Returns whether this queue is awaiting a final event to be delivered.
    boolean
    isClosed()
    Checks if this queue has been closed.
    abstract void
    signalQueuePollerStarted()
    Signals that the queue poller has started consuming events.
    abstract int
    size()
    Returns the current size of the queue.
    abstract EventQueue
    tap()
    Creates a child queue that shares events with this queue.
    void
    taskDone()
    Placeholder method for task completion notification.

    Methods inherited from class java.lang.Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • DEFAULT_QUEUE_SIZE

      public static final int DEFAULT_QUEUE_SIZE
      Default maximum queue size for event queues.
      See Also:

  • Constructor Details

    • EventQueue

      protected EventQueue()
      Creates an EventQueue with the default queue size.

    • EventQueue

      protected EventQueue(int queueSize)
      Creates an EventQueue with the specified queue size.
      Parameters:
      queueSize - the maximum number of events that can be queued
      Throws:
      IllegalArgumentException - if queueSize is less than or equal to 0

    • EventQueue

      protected EventQueue(EventQueue parent)
      Creates an EventQueue as a child of the specified parent queue.
      Parameters:
      parent - the parent event queue

  • Method Details

    • getQueueSize

      public int getQueueSize()
      Returns the configured queue size.
      Returns:
      the maximum number of events that can be queued

    • awaitQueuePollerStart

      public abstract void awaitQueuePollerStart() throws InterruptedException
      Waits for the queue poller to start consuming events. This method blocks until signaled by signalQueuePollerStarted().
      Throws:
      InterruptedException - if the thread is interrupted while waiting

    • signalQueuePollerStarted

      public abstract void signalQueuePollerStarted()
      Signals that the queue poller has started consuming events. This unblocks any threads waiting in awaitQueuePollerStart().

    • enqueueEvent

      public void enqueueEvent(Event event)
      Enqueues an event for processing.
      Parameters:
      event - the event to enqueue

    • enqueueItem

      public abstract void enqueueItem(EventQueueItem item)
      Enqueues an event queue item for processing.

      This method will block if the queue is full, waiting to acquire a semaphore permit. If the queue is closed, the event will not be enqueued and a warning will be logged.

      Parameters:
      item - the event queue item to enqueue
      Throws:
      RuntimeException - if interrupted while waiting to acquire the semaphore

    • enqueueLocalOnly

      public void enqueueLocalOnly(EventQueueItem item)
      Enqueues an event directly to this specific queue only, bypassing the MainEventBus.

      This method is used for enqueuing already-persisted events (e.g., current task state on subscribe) that should only be sent to this specific subscriber, not distributed to all children or sent through MainEventBusProcessor.

      Default implementation throws UnsupportedOperationException. Only ChildQueue supports this.

      Parameters:
      item - the event queue item to enqueue directly
      Throws:
      UnsupportedOperationException - if called on MainQueue or other queue types

    • enqueueEventLocalOnly

      public void enqueueEventLocalOnly(Event event)
      Enqueues an event directly to this specific queue only, bypassing the MainEventBus.

      Convenience method that wraps the event in a LocalEventQueueItem before calling enqueueLocalOnly(EventQueueItem).

      Parameters:
      event - the event to enqueue directly
      Throws:
      UnsupportedOperationException - if called on MainQueue or other queue types

    • tap

      public abstract EventQueue tap()
      Creates a child queue that shares events with this queue.

      For MainQueue: creates a ChildQueue that receives all events enqueued to the parent. For ChildQueue: throws IllegalStateException (only MainQueue can be tapped).

      Returns:
      a new ChildQueue instance
      Throws:
      IllegalStateException - if called on a ChildQueue

    • dequeueEventItem

      public abstract @Nullable EventQueueItem dequeueEventItem(int waitMilliSeconds) throws EventQueueClosedException
      Dequeues an EventQueueItem from the queue.

      This method returns the full EventQueueItem wrapper, allowing callers to check metadata like whether the event is replicated via EventQueueItem.isReplicated().

      Note: MainQueue does not support dequeue operations - only ChildQueues can be consumed.

      Parameters:
      waitMilliSeconds - the maximum time to wait in milliseconds
      Returns:
      the EventQueueItem, or null if timeout occurs
      Throws:
      EventQueueClosedException - if the queue is closed and empty
      UnsupportedOperationException - if called on MainQueue

    • taskDone

      public void taskDone()
      Placeholder method for task completion notification. Currently not used as BlockingQueue.poll()/take() automatically remove events.

    • size

      public abstract int size()
      Returns the current size of the queue.

      For MainQueue: returns the number of events in-flight (in MainEventBus queue + currently being processed). This reflects actual capacity usage tracked by the semaphore. For ChildQueue: returns the size of the local consumption queue.

      Returns:
      the number of events currently in the queue

    • isAwaitingFinalEvent

      public boolean isAwaitingFinalEvent()
      Returns whether this queue is awaiting a final event to be delivered.

      This is used by EventConsumer to determine if it should keep polling even when the queue is empty. A final event may still be in-transit through MainEventBusProcessor.

      For MainQueue: always returns false (MainQueue cannot be consumed). For ChildQueue: returns true if EventQueue.ChildQueue.expectFinalEvent() was called but the final event hasn't been received yet.

      Returns:
      true if awaiting a final event, false otherwise

    • clearAwaitingFinalEvent

      public void clearAwaitingFinalEvent()
      Clears the awaiting final event flag.

      Default implementation is a no-op for queues that don't track this state. ChildQueue overrides this to actually clear the flag.

    • close

      public abstract void close()
      Closes this event queue gracefully, allowing pending events to be consumed.
      Specified by:
      close in interface AutoCloseable

    • close

      public abstract void close(boolean immediate)
      Closes this event queue with control over immediate shutdown.
      Parameters:
      immediate - if true, clears all pending events immediately; if false, allows graceful drain

    • close

      public abstract void close(boolean immediate, boolean notifyParent)
      Close this queue with control over parent notification (ChildQueue only).
      Parameters:
      immediate - If true, clear all pending events immediately
      notifyParent - If true, notify parent (standard behavior). If false, close this queue without decrementing parent's reference count (used for non-blocking non-final tasks to keep MainQueue alive for resubscription)
      Throws:
      UnsupportedOperationException - if called on MainQueue

    • isClosed

      public boolean isClosed()
      Checks if this queue has been closed.
      Returns:
      true if the queue is closed, false otherwise

    • doClose

      protected void doClose()
      Internal method to close the queue gracefully. Delegates to doClose(boolean) with immediate=false.

    • doClose

      protected void doClose(boolean immediate)
      Internal method to close the queue with control over immediate shutdown.
      Parameters:
      immediate - if true, clears all pending events immediately; if false, allows graceful drain