irisandClaude Sonnet
e8dbcaa7db
Adds sse.rs (a pure port of Sse.kt's frame parser), api.rs (a Transport trait plus a ureq-backed implementation and an ApiClient covering the session lifecycle: list/read, message/unqueue/answer, interrupt/stop/start, title/cwd/model/permission-mode/notify, command/compact, delete, and a transcript page), and event_stream.rs (follow_session_events, mirroring EventStream.kt's reset/event split). ureq rather than reqwest: server/ already depends on it for its own outbound HTTPS, this stays blocking like Api.kt's HttpURLConnection calls with no async runtime to carry, and its own PEM cert support means no extra rustls/rustls-pemfile dependency to pin. Network I/O sits behind Transport so ApiClient and follow_session_events are tested with fakes, no server involved. Not yet covered, tracked in CLIENT_CORE.md: setups, the file explorer, usage, models, and attachments/import. cargo test (78 passed), clippy --all-targets and fmt clean. Co-Authored-By: Claude Sonnet <noreply@anthropic.com>
143 lines
5.1 KiB
Rust
143 lines
5.1 KiB
Rust
//! The SSE half of the API: one long-lived GET per open session screen,
|
|
//! replaying the transcript after a cursor and then following it live.
|
|
//! Ported from `app/.../EventStream.kt`; the framing itself is
|
|
//! [`crate::sse`].
|
|
|
|
use std::io::{BufRead, BufReader};
|
|
|
|
use event_model::SeqEvent;
|
|
|
|
use crate::api::{ApiError, Transport};
|
|
use crate::sse::SseReader;
|
|
|
|
/// The frame name the server uses to say a cursor was too far behind to
|
|
/// continue from. Must match `send_backlog` in `server/src/routes.rs`.
|
|
const RESET_EVENT: &str = "reset";
|
|
|
|
/// One frame of a session's event stream, folded from the wire shape the
|
|
/// caller needs to act on -- mirroring what `EventStream.kt`'s three
|
|
/// callbacks were for, as a single enum instead, since Rust has no
|
|
/// equivalent of handing three closures to one blocking call.
|
|
pub enum StreamItem {
|
|
/// The connection was accepted; the measured moment the stream is live
|
|
/// (see `EventStream.kt`'s doc on `onOpen` for why this, not the first
|
|
/// event, is what clears a previous failure on screen).
|
|
Open,
|
|
/// The cursor was too far behind to continue from: everything already
|
|
/// displayed is stale, and the events that follow are a fresh window.
|
|
/// Arrives before those events, so a caller that clears on it stays in
|
|
/// order.
|
|
Reset,
|
|
/// One event, as both the raw line the transcript cache stores and the
|
|
/// parsed [`SeqEvent`] the fold works from -- they have to be the same
|
|
/// line, so both travel together rather than being parsed twice from
|
|
/// two call sites.
|
|
Event { raw: String, event: SeqEvent },
|
|
}
|
|
|
|
/// Follows `/sessions/{id}/events?after={after}`, calling `on_item` for
|
|
/// each [`StreamItem`] until the connection drops or `on_item` asks to
|
|
/// stop (by returning `false`). Reconnecting -- with the last seq seen as
|
|
/// the new cursor -- is the caller's job, same as in the Kotlin version.
|
|
pub fn follow_session_events(
|
|
transport: &dyn Transport,
|
|
session_id: &str,
|
|
after: u64,
|
|
mut on_item: impl FnMut(StreamItem) -> bool,
|
|
) -> Result<(), ApiError> {
|
|
let path = format!("/sessions/{session_id}/events?after={after}");
|
|
let body = transport.stream(&path)?;
|
|
if !on_item(StreamItem::Open) {
|
|
return Ok(());
|
|
}
|
|
let mut lines = BufReader::new(body).lines();
|
|
let mut reader = SseReader::new();
|
|
while let Some(line) = lines.next().transpose().map_err(|e| ApiError {
|
|
message: format!("Can't reach the server -- retrying. ({e})"),
|
|
status: None,
|
|
})? {
|
|
let Some(frame) = reader.feed_line(&line) else {
|
|
continue;
|
|
};
|
|
// A named frame carries no payload and a data frame has no name.
|
|
if frame.name.as_deref() == Some(RESET_EVENT) {
|
|
if !on_item(StreamItem::Reset) {
|
|
return Ok(());
|
|
}
|
|
} else if !frame.data.is_empty() {
|
|
let event: SeqEvent = serde_json::from_str(&frame.data).map_err(|e| ApiError {
|
|
message: format!("The server sent an event this build couldn't parse: {e}"),
|
|
status: None,
|
|
})?;
|
|
if !on_item(StreamItem::Event {
|
|
raw: frame.data,
|
|
event,
|
|
}) {
|
|
return Ok(());
|
|
}
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::api::{Body, RawResponse};
|
|
use std::io::Cursor;
|
|
|
|
struct FixtureTransport {
|
|
body: &'static str,
|
|
}
|
|
|
|
impl Transport for FixtureTransport {
|
|
fn request(
|
|
&self,
|
|
_method: &str,
|
|
_path: &str,
|
|
_body: Option<Body>,
|
|
) -> Result<RawResponse, ApiError> {
|
|
unimplemented!("this fixture only serves a stream")
|
|
}
|
|
|
|
fn stream(&self, _path: &str) -> Result<Box<dyn std::io::Read + Send>, ApiError> {
|
|
Ok(Box::new(Cursor::new(self.body.as_bytes().to_vec())))
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn events_and_a_reset_frame_are_told_apart() {
|
|
let transport = FixtureTransport {
|
|
body: "event:reset\n\ndata:{\"seq\":1,\"ts\":1.0,\"type\":\"status\",\"state\":\"idle\"}\n\n",
|
|
};
|
|
let mut items = Vec::new();
|
|
follow_session_events(&transport, "s1", 0, |item| {
|
|
items.push(match item {
|
|
StreamItem::Open => "open".to_string(),
|
|
StreamItem::Reset => "reset".to_string(),
|
|
StreamItem::Event { event, .. } => format!("event:{}", event.seq),
|
|
});
|
|
true
|
|
})
|
|
.unwrap();
|
|
assert_eq!(items, vec!["open", "reset", "event:1"]);
|
|
}
|
|
|
|
#[test]
|
|
fn the_caller_can_stop_early() {
|
|
let transport = FixtureTransport {
|
|
body: "data:{\"seq\":1,\"ts\":1.0,\"type\":\"status\",\"state\":\"idle\"}\n\n\
|
|
data:{\"seq\":2,\"ts\":1.0,\"type\":\"status\",\"state\":\"idle\"}\n\n",
|
|
};
|
|
let mut count = 0;
|
|
follow_session_events(&transport, "s1", 0, |item| {
|
|
if matches!(item, StreamItem::Event { .. }) {
|
|
count += 1;
|
|
}
|
|
count < 1
|
|
})
|
|
.unwrap();
|
|
assert_eq!(count, 1);
|
|
}
|
|
}
|