Struct EventStream 

pub struct EventStream { /* private fields */ }

An async stream of StreamResponse events from an SSE endpoint.

Created by crate::A2aClient::stream_message or crate::A2aClient::subscribe_to_task. Call EventStream::next in a loop to consume events.

When dropped, the background body-reader task is aborted to prevent resource leaks.

Implementations

impl EventStream

pub fn from_event_channel( rx: Receiver<Result<StreamResponse, ClientError>>, ) -> EventStream

Creates an EventStream from a channel of already-decoded events.

This is the constructor an out-of-tree crate::transport::Transport needs. Transport::send_streaming_request must return an EventStream, and every other way to build one is pub(crate) — so before this existed, a custom transport could implement the unary half of the trait and not the streaming half. A binding crate cannot be written against half a trait, so this is the piece that makes the extension point whole.

Feed rx from a background task that decodes the transport’s own frames into StreamResponse values. Sending Err delivers that error to the consumer and is the right way to report a decode failure mid-stream — silently ending the stream would be indistinguishable, to the consumer, from the task finishing normally.

Dropping the sender ends the stream. As on every binding, that is a completion only after a final event (a Message, or a terminal or interrupted task state); before one, the consumer receives ClientError::IncompleteStream.

The returned stream aborts the bridging task when dropped, exactly as the built-in transports’ streams do.

Example

let (tx, rx) = tokio::sync::mpsc::channel(64);
tokio::spawn(async move {
    while let Some(frame) = my_transport_stream.next().await {
        if tx.send(frame.map(into_stream_response)).await.is_err() {
            break; // consumer dropped the stream
        }
    }
});
Ok(EventStream::from_event_channel(rx))

pub const fn with_first_event_timeout(self, timeout: Duration) -> EventStream

Bounds the wait for the first chunk of stream data.

If no data arrives within timeout of the stream being created, EventStream::next yields a ClientError::Timeout instead of blocking forever. Once any data is received this bound is spent and with_idle_timeout governs instead.

Wired by every streaming transport (JSON-RPC, REST, gRPC, WebSocket): their connect timeouts only bound establishment, and a server that establishes a stream and then goes silent must not hang the consumer. A2aClient then replaces the transport’s value with ClientConfig::stream_first_event_timeout on every stream it returns.

pub const fn with_idle_timeout(self, timeout: Option<Duration>) -> EventStream

Bounds the silence between chunks once the first has arrived; None removes the bound.

Defaults to DEFAULT_STREAM_IDLE_TIMEOUT. Any bytes reset it, so an SSE server that sends keep-alive comments keeps a quiet stream open indefinitely. When it expires, next yields ClientError::Timeout once and the stream ends; the background reader is aborted, which closes the connection.

A2aClient sets this from ClientConfig::stream_idle_timeout on every stream it returns. Call it yourself on a stream obtained from a Transport directly.

pub const fn with_max_event_size(self, max_bytes: usize) -> EventStream

Sets the largest single event this stream accepts, in bytes. An event over it is refused with ClientError::Transport and skipped; see ClientConfig::max_event_size, which A2aClient applies to every stream it returns.

pub const fn status_code(&self) -> u16

Returns the HTTP status code from the response that established this stream.

The transport layer validates the HTTP status during stream establishment and returns an error for non-2xx responses, so this is typically 200.

pub fn last_event_id(&self) -> Option<&str>

The SSE id: of the last event received, or None if none carried one.

Pass it to A2aClient::subscribe_to_task_from after a disconnect and a server that keeps an event log (this repository’s does) replays what was missed. It follows SSE’s lastEventId rules: an id persists across events that carry none, so a server-synthesized frame without an id does not move it. Always None on gRPC and WebSocket streams, whose frames carry no ids.

pub async fn next(&mut self) -> Option<Result<StreamResponse, ClientError>>

Returns the next event from the stream.

Returns None when the stream has finished: after a terminal status update, or when the body ends after an event the stream may end on (a Message, or a task or status update in a terminal or interrupted state).

Returns Some(Err(...)) on transport or protocol errors, including ClientError::IncompleteStream when the body ends anywhere else, and ClientError::Timeout when a silence bound expires. After an error the stream is finished and the next call returns None.

Trait Implementations

impl Debug for EventStream

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl Drop for EventStream

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.