Type Alias SseChatStream 

pub type SseChatStream<S, D> = FramedChatStream<SseStream<S>, D>;

A chat stream decoded from a response’s SSE events by an adapter’s ChatFrameDecoder<SseEvent>: the FramedChatStream over an SseStream, with the driver guarantees the decoder trait lists.

SseChatStream::new classifies a failure with SseError::into_error, for a body whose error converts into a TransportError, as an HttpResponse body’s does; with_error_map takes any other body, with a map that keeps the rules of into_error: a timeout stays a timeout, invalid UTF-8 is a stream decode error, an event past the size cap a stream protocol error, and the messages are the adapter’s own.

It is unboxed and Send (for a Send body); box it into a ChatStream with Prepared::finish_stream:

use allama::{Error, Result, StreamEvent, StreamErrorKind};
use allama_http::TransportError;
use allama_http::framing::ChatFrameDecoder;
use allama_http::sse::{SseChatStream, SseEvent, SseStream};
use futures_util::{FutureExt, StreamExt, stream};

/// `data:` text deltas, then `event: done`.
struct Lines {
    started: bool,
    done: bool,
}

impl ChatFrameDecoder<SseEvent> for Lines {
    fn frame(&mut self, event: SseEvent) -> Vec<Result<StreamEvent>> {
        let mut out = Vec::new();
        if !self.started {
            self.started = true;
            out.push(Ok(StreamEvent::response_start(None, None)));
            out.push(Ok(StreamEvent::block_start(0, allama::StreamBlockType::Text)));
        }
        if event.event == "done" {
            self.done = true;
            out.push(Ok(StreamEvent::block_stop(0)));
            out.push(Ok(StreamEvent::ResponseStop));
        } else {
            out.push(Ok(StreamEvent::text_delta(0, event.data)));
        }
        out
    }

    fn end(&mut self) -> Vec<Result<StreamEvent>> {
        let error = Error::stream(StreamErrorKind::Incomplete, "the stream ended early");
        vec![Err(error.authored().with_provider("lines"))]
    }

    fn fail(&mut self, error: Error) -> Vec<Result<StreamEvent>> {
        vec![Err(error.with_provider("lines"))]
    }

    fn is_done(&self) -> bool {
        self.done
    }
}

let body = stream::iter([Ok::<_, TransportError>(
    &b"data: Hel\n\ndata: lo\n\nevent: done\ndata: {}\n\n"[..],
)]);
let stream = SseChatStream::new(SseStream::new(body), Lines { started: false, done: false });
let events: Vec<_> = stream.collect().now_or_never().unwrap();
assert_eq!(events.len(), 6);
assert!(matches!(events[5], Ok(StreamEvent::ResponseStop)));

Aliased Type

pub struct SseChatStream<S, D> { /* private fields */ }