Trait ChatFrameDecoder 

pub trait ChatFrameDecoder<F>: Send + 'static {
    // Required methods
    fn frame(&mut self, frame: F) -> Vec<Result<StreamEvent>> ;
    fn end(&mut self) -> Vec<Result<StreamEvent>> ;

    // Provided methods
    fn fail(&mut self, error: Error) -> Vec<Result<StreamEvent>>  { ... }
    fn is_done(&self) -> bool { ... }
}

An adapter’s decoder for one streamed response: frames of type F in (an SseEvent, an NDJSON line), chat stream events out. FramedChatStream drives it.

The decoder holds what only the adapter knows: its wire format, when its response starts and ends, and its error classification. The driver holds the loop every streaming adapter needs, and its guarantees:

• each frame goes to frame, in order;
• the stream ends after the first error, from the decoder or from the body: the items after it are dropped;
• when the body ends, end runs; when reading or framing it fails, fail runs, with the failure already classified. At most one of them runs, once, and only while the decoder is not done (it has yielded no error and is_done is false);
• the stream ends once the decoder is done, without waiting for the body to close, and the body is never polled after the stream ended.

A decoder yields warnings and response-level provider data as they arrive, even before its ResponseStart: Prepared::finish_stream puts them after the start (start_first).

An SSE adapter implements ChatFrameDecoder<SseEvent>, and SseChatStream drives it.

Each method returns the items of one frame as a Vec, and this shape is deliberate and stable. An empty Vec does not allocate, and a frame that yields events was decoded from JSON anyway, so an out: &mut Vec<_> sink parameter would save one small allocation per frame and cost every decoder a harder signature: early returns become pushes, and a decoder that forgets to push loses an error silently.

use allama::{Error, Result, StreamEvent, StreamErrorKind};
use allama_http::TransportError;
use allama_http::framing::{ChatFrameDecoder, FramedChatStream};
use allama_http::ndjson::NdjsonStream;
use futures_util::{FutureExt, StreamExt, stream};

/// `{"text": ..}` lines, then `{"done": true}`.
struct Lines {
    started: bool,
    done: bool,
}

impl ChatFrameDecoder<String> for Lines {
    fn frame(&mut self, line: String) -> Vec<Result<StreamEvent>> {
        let json: serde_json::Value = match serde_json::from_str(&line) {
            Ok(json) => json,
            Err(error) => return vec![Err(Error::stream_decode(error.to_string()))],
        };
        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 json["done"] == true {
            self.done = true;
            out.push(Ok(StreamEvent::block_stop(0)));
            out.push(Ok(StreamEvent::ResponseStop));
        } else {
            out.push(Ok(StreamEvent::text_delta(0, json["text"].as_str().unwrap_or_default())));
        }
        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"{\"text\":\"Hel\"}\n{\"text\":\"lo\"}\n{\"done\":true}\n"[..],
)]);
let lines = NdjsonStream::new(body);
let stream = FramedChatStream::new(lines, 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)));

Required Methods

fn frame(&mut self, frame: F) -> Vec<Result<StreamEvent>>

Decode one frame. An error among the items is the stream’s last item: the driver drops the items after it and ends the stream.

fn end(&mut self) -> Vec<Result<StreamEvent>>

The body ended before the decoder was done: what the stream still yields, typically a StreamErrorKind::Incomplete error, since the terminal frame never arrived. The stream ends after it, whatever it returns, nothing included.

Provided Methods

fn fail(&mut self, error: Error) -> Vec<Result<StreamEvent>>

Reading or framing the body failed before the decoder was done. error is the classified failure (the framer’s error converted into an allama::Error, or the driver’s error map). The default yields it as it is; override it to close open blocks first, to attach the provider and the request id, or to add what the adapter knows about where to resume. The stream ends after it.

fn is_done(&self) -> bool

Whether the decoder is done: its terminal frame arrived. The driver then ends the stream without waiting for the body to close, and calls neither end nor fail. The default is false: the stream ends with the body.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors