Package org.a2aproject.sdk.server.events

Interface TaskStreamLifecycleHook

All Known Implementing Classes:
CloseStreamsHook, DefaultTaskStreamLifecycleHook
public interface TaskStreamLifecycleHook
Hook for observing task stream lifecycle events and controlling stream resources.

Implementations are notified when clients subscribe/unsubscribe to a task's event stream and when events are processed for a task. The StreamCloseHandle passed to each callback can be used to gracefully close all active streams for the task.

Ordering guarantees

  • onSubscribe is called synchronously during MainQueue.tap(), after the ChildQueue has been added to the children list but before the ChildQueue is returned to the caller. The subscriber count visible via StreamCloseHandle.getActiveSubscriberCount() includes the new subscriber.
  • onEvent is called on the MainEventBusProcessor thread after the event has been persisted and distributed to all ChildQueues. Implementations must return promptly to avoid stalling event distribution for other tasks.
  • Because onSubscribe and onEvent run on different threads, a fast event emission may cause onEvent to fire concurrently with or even before onSubscribe returns. Stateful hook implementations must be thread-safe.
  • onUnsubscribe is called synchronously inside ChildQueue.close(), on whichever thread closes the child (EventConsumer, hook via StreamCloseHandle.closeStreams(), or the client's transport layer).

Hook binding lifetime

The hook is set on a MainQueue when it is first created (in InMemoryQueueManager.createOrTap()). If the MainQueue already exists (e.g., a second client subscribes to the same task), the existing hook reference is retained. The hook instance is effectively bound for the lifetime of the MainQueue.

The default implementation is a no-op. To provide custom behavior (e.g., closing streams after a timeout), implement this interface and register it as a CDI alternative: 


 @ApplicationScoped
 @Alternative
 @Priority(1)
 public class MyStreamHook implements TaskStreamLifecycleHook {
     // ...
 }
See Also:

  • Method Summary

    Modifier and Type
    Method
    Description
    void
    onEvent(String taskId, Event event, StreamCloseHandle handle)
    Called after an event has been persisted and distributed to all ChildQueues.
    void
    onSubscribe(String taskId, StreamCloseHandle handle)
    Called when a new ChildQueue is created for a task (a client subscribes to the stream).
    void
    onUnsubscribe(String taskId, StreamCloseHandle handle)
    Called when a ChildQueue closes for a task (a client disconnects or streams are closed).

  • Method Details

    • onSubscribe

      void onSubscribe(String taskId, StreamCloseHandle handle)
      Called when a new ChildQueue is created for a task (a client subscribes to the stream).

      If this method calls StreamCloseHandle.closeStreams(), the ChildQueue being created will be closed before tap() returns it to the caller. The caller will receive a closed queue whose dequeueEventItem() throws EventQueueClosedException immediately.

      Parameters:
      taskId - the task identifier
      handle - handle to close streams and query subscriber count

    • onUnsubscribe

      void onUnsubscribe(String taskId, StreamCloseHandle handle)
      Called when a ChildQueue closes for a task (a client disconnects or streams are closed).

      Calling StreamCloseHandle.closeStreams() from within this callback is safe but has no effect — a reentrancy guard prevents recursive iteration over the children list. To close remaining streams in response to an unsubscription, schedule the call asynchronously or handle it in onEvent(java.lang.String, org.a2aproject.sdk.spec.Event, org.a2aproject.sdk.server.events.StreamCloseHandle).

      Parameters:
      taskId - the task identifier
      handle - handle to close streams and query subscriber count

    • onEvent

      void onEvent(String taskId, Event event, StreamCloseHandle handle)
      Called after an event has been persisted and distributed to all ChildQueues.
      Parameters:
      taskId - the task identifier
      event - the event that was processed
      handle - handle to close streams and query subscriber count