Struct SseStream 

pub struct SseStream<T> { /* private fields */ }

A type-safe wrapper around Server-Sent Events (SSE) streams

Leverages reqwest-sse for SSE protocol parsing and adds:

• Automatic JSON deserialization to typed structs
• Stream terminator support (e.g., [DONE] for OpenAI-style APIs)
• Integrated error handling with ApiError
• Content-Type validation (text/event-stream required)

Charset Handling

The reqwest-sse library automatically handles charset detection and decoding based on the Content-Type header. If no charset is specified, UTF-8 is assumed. This matches the SSE specification default behavior.

Example

Basic usage with async iteration:

use futures::StreamExt;

let stream: SseStream<CompletionChunk> = client.stream_completions(request).await?;
let mut stream = std::pin::pin!(stream);

while let Some(result) = stream.next().await {
    match result {
        Ok(chunk) => println!("Received: {:?}", chunk),
        Err(e) => eprintln!("Error: {}", e),
    }
}

Error Handling

The stream yields Result<T, ApiError> items. Errors can occur from:

• Invalid JSON in SSE data field (ApiError::Serialization)
• SSE protocol errors (ApiError::SseParseError)
• Network errors during streaming

Important: When an error occurs for a single event (e.g., malformed JSON), the stream yields Err for that item but continues streaming subsequent events. The stream only ends when:

• A terminator is received (if configured)
• The server closes the connection
• A fatal network error occurs

This allows the client to handle per-event errors gracefully without losing the entire stream. Compare this to other error handling strategies where a single bad event might terminate the stream.

Terminator Support

When a terminator string is specified (e.g., [DONE]), the stream automatically ends when an SSE event with that exact data is received. The terminator event itself is not yielded to the consumer.

Implementations

impl<T> SseStream<T>

where T: DeserializeOwned,

pub fn with_metadata(self) -> SseStreamWithMetadata<T>

Convert this stream into one that yields events with metadata

This consumes the stream and returns a new stream that yields SseEvent<T> instead of just T, providing access to SSE metadata (event type, id, retry).

Example

use futures::StreamExt;

let stream = client.stream_completions(request).await?;
let mut stream_with_metadata = stream.with_metadata();
let mut stream_with_metadata = std::pin::pin!(stream_with_metadata);

while let Some(result) = stream_with_metadata.next().await {
    match result {
        Ok(event) => {
            println!("Data: {:?}", event.data);
            println!("Event type: {}", event.metadata.event);
            if let Some(id) = &event.metadata.id {
                println!("Event ID: {}", id);
            }
        }
        Err(e) => eprintln!("Error: {}", e),
    }
}

Trait Implementations

impl<T> Stream for SseStream<T>

where T: DeserializeOwned,

type Item = Result<T, ApiError>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.

impl<'pin, T> Unpin for SseStream<T>

where PinnedFieldsOf<__SseStream<'pin, T>>: Unpin,