Struct StreamingFanout 

pub struct StreamingFanout<L> { /* private fields */ }

Fan-out of task updates to live readers and callback subscribers, over an AsyncEventLog that assigns the ids and keeps the events for replay.

Cloning shares the underlying per-task state (an Arc<Mutex<…>>) and the log, so a clone observes the same channels and subscribers.

Implementations

impl<L> StreamingFanout<L>

pub fn over(log: L) -> Self

Build a fan-out over log.

The log is what a client resumes from, so pass a durable one where resumption has to survive a restart.

impl StreamingFanout<InMemoryEventLog>

pub fn new() -> Self

Create a handler over a fresh in-memory log.

Trait Implementations

impl<L: AsyncEventLog + Clone + 'static> AsyncStreamingHandler for StreamingFanout<L>

fn remove_task_subscribers<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Drop the task’s live readers and callback subscribers.

The event log is left alone: a task with nobody listening is exactly the case a client is about to resume from. Deleting the events is AsyncEventLog::discard, which an operator’s retention sweep calls.

fn add_status_subscriber<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, subscriber: Box<dyn Subscriber<TaskStatusUpdateEvent> + Send + Sync>, ) -> Pin<Box<dyn Future<Output = Result<String, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Add a status update subscriber for a task

fn add_artifact_subscriber<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, subscriber: Box<dyn Subscriber<TaskArtifactUpdateEvent> + Send + Sync>, ) -> Pin<Box<dyn Future<Output = Result<String, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Add an artifact update subscriber for a task

fn remove_subscription<'life0, 'life1, 'async_trait>( &'life0 self, _subscription_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Remove a specific subscription

fn get_subscriber_count<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<usize, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get the number of active subscribers for a task

fn broadcast_status_update<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, update: TaskStatusUpdateEvent, ) -> Pin<Box<dyn Future<Output = Result<(), A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Broadcast a status update to all subscribers of a task

fn broadcast_artifact_update<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, update: TaskArtifactUpdateEvent, ) -> Pin<Box<dyn Future<Output = Result<(), A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Broadcast an artifact update to all subscribers of a task

fn status_update_stream<'life0, 'life1, 'async_trait>( &'life0 self, _task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<TaskStatusUpdateEvent, A2AError>> + Send>>, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create a stream of status updates for a task

fn artifact_update_stream<'life0, 'life1, 'async_trait>( &'life0 self, _task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<TaskArtifactUpdateEvent, A2AError>> + Send>>, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create a stream of artifact updates for a task

fn combined_update_stream<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, from_event_id: Option<u64>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<SeqEvent, A2AError>> + Send>>, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create a combined stream of all updates for a task.

fn has_subscribers<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Check if a task has any active subscribers

fn validate_streaming_params<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Validate streaming parameters

fn start_task_streaming<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, from_event_id: Option<u64>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<SeqEvent, A2AError>> + Send>>, A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Start streaming for a task with automatic cleanup.

fn stop_task_streaming<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), A2AError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Stop all streaming for a task

impl<L: Clone> Clone for StreamingFanout<L>

fn clone(&self) -> StreamingFanout<L>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Default for StreamingFanout<InMemoryEventLog>

fn default() -> Self

Returns the “default value” for a type.