diff --git a/client-core/Cargo.lock b/client-core/Cargo.lock index 91df4a0..42f6d46 100644 --- a/client-core/Cargo.lock +++ b/client-core/Cargo.lock @@ -47,8 +47,6 @@ name = "client-core" version = "0.1.0" dependencies = [ "event-model", - "rustls", - "rustls-pemfile", "serde", "serde_json", "tempfile", @@ -484,15 +482,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "rustls-pemfile" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" -dependencies = [ - "rustls-pki-types", -] - [[package]] name = "rustls-pki-types" version = "1.15.1" diff --git a/client-core/Cargo.toml b/client-core/Cargo.toml index 1523261..1560b81 100644 --- a/client-core/Cargo.toml +++ b/client-core/Cargo.toml @@ -27,12 +27,6 @@ serde_json = { version = "1", features = ["float_roundtrip"] } # no need of an async runtime, and RUST.md's brief for this port is # "lightweight" throughout. ureq = { version = "3", features = ["json"] } -# Verifying the server's pinned self-signed leaf against the CA `wg-app-link` -# mints, the same way `ServerConfig.kt`'s `applyPinnedTls` does. rustls -# rather than native-tls because ureq is already rustls-backed here and the -# rest of this project's TLS goes through rustls too. -rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } -rustls-pemfile = "2" [dev-dependencies] tempfile = "3" diff --git a/client-core/src/api.rs b/client-core/src/api.rs new file mode 100644 index 0000000..72fd047 --- /dev/null +++ b/client-core/src/api.rs @@ -0,0 +1,543 @@ +//! The REST half of the backend's surface (see `server/src/routes.rs`'s +//! module doc for the table); the SSE half is [`crate::event_stream`]. +//! Ported from `app/.../Api.kt`, but **not at full parity yet** -- see +//! `CLIENT_CORE.md` for exactly which routes have a typed method here and +//! which do not. +//! +//! Network I/O sits behind the [`Transport`] trait so the rest of this +//! crate, and anything built on it, can be tested against a fake one with +//! no server involved. [`UreqTransport`] is the only real implementation. + +use std::io::Read; + +use serde::Deserialize; +use serde_json::Value; + +/// A request that did not produce what it asked for, carrying the server's +/// own wording where it sent some. +/// +/// `status` is the HTTP status where there was a response at all, and +/// `None` where the server was never reached -- mirroring `ApiException` in +/// `Api.kt`. +#[derive(Debug, Clone)] +pub struct ApiError { + pub message: String, + pub status: Option, +} + +impl std::fmt::Display for ApiError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.message) + } +} +impl std::error::Error for ApiError {} + +/// A request body to send, in whichever of the two shapes the surface +/// takes: `Api.kt`'s `jsonBody` and `streamBody`. +pub enum Body { + Json(Value), + Bytes { + content_type: String, + bytes: Vec, + }, +} + +/// What a transport hands back for a REST call: the status and the body +/// read whole. A streamed body ([`Transport::stream`]) is a different +/// method because its whole point is not reading it whole. +pub struct RawResponse { + pub status: u16, + pub body: Vec, +} + +/// The network boundary this crate's pure logic is kept out from behind. +/// `server/src/routes.rs`'s module doc is the surface this drives. +pub trait Transport: Send + Sync { + /// One request/response call -- everything but the long-lived SSE GETs. + fn request( + &self, + method: &str, + path: &str, + body: Option, + ) -> Result; + + /// Opens `path` and answers a reader over the response body, for a + /// caller that reads it as a stream rather than all at once (the SSE + /// connections in [`crate::event_stream`]). Fails the same way + /// [`Transport::request`] does for a non-2xx response. + fn stream(&self, path: &str) -> Result, ApiError>; +} + +/// One session as `GET /sessions` and `GET /sessions/{id}` report it. +/// Mirrors `Api.kt`'s `SessionSummary`; see that type's doc for what each +/// field means and why `setup` is never shown. +#[derive(Debug, Clone, PartialEq, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionSummary { + pub id: String, + pub setup: String, + #[serde(default)] + pub keeps_own_transcript: bool, + pub setup_name: String, + pub provider: String, + pub title: String, + #[serde(default)] + pub model: Option, + #[serde(default)] + pub permission_mode: Option, + #[serde(default)] + pub imported: bool, + #[serde(default = "default_true")] + pub notify: bool, + #[serde(default)] + pub cwd: Option, + #[serde(default)] + pub context_tokens: Option, + #[serde(default)] + pub max_image_edge: Option, + pub status: String, + pub last_activity: f64, +} + +fn default_true() -> bool { + true +} + +/// A client-core equivalent of `requestFromServer` plus the typed calls +/// built on it. Holds no state of its own beyond the transport -- the +/// session id or setup id a call is about is a parameter, per this +/// project's "ask for the least you need". +pub struct ApiClient { + transport: T, +} + +impl ApiClient { + pub fn new(transport: T) -> Self { + Self { transport } + } + + fn json_request Deserialize<'de>>( + &self, + method: &str, + path: &str, + body: Option, + ) -> Result { + let raw = self.transport.request(method, path, body.map(Body::Json))?; + serde_json::from_slice(&raw.body).map_err(|e| ApiError { + message: format!("Reached the server but couldn't read its response ({e})"), + status: Some(raw.status), + }) + } + + fn empty_request(&self, method: &str, path: &str, body: Option) -> Result<(), ApiError> { + self.transport.request(method, path, body.map(Body::Json))?; + Ok(()) + } + + pub fn fetch_sessions(&self) -> Result, ApiError> { + self.json_request("GET", "/sessions", None) + } + + pub fn fetch_session(&self, session_id: &str) -> Result { + self.json_request("GET", &format!("/sessions/{session_id}"), None) + } + + pub fn send_message( + &self, + session_id: &str, + text: &str, + attachment_ids: &[String], + ) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/message"), + Some(serde_json::json!({ "text": text, "attachmentIds": attachment_ids })), + ) + } + + pub fn unqueue_message(&self, session_id: &str, message_id: &str) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/unqueue"), + Some(serde_json::json!({ "messageId": message_id })), + ) + } + + pub fn answer_question( + &self, + session_id: &str, + question_id: &str, + answers: &[String], + ) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/answer"), + Some(serde_json::json!({ "questionId": question_id, "answers": answers })), + ) + } + + pub fn interrupt_session(&self, session_id: &str) -> Result<(), ApiError> { + self.empty_request("POST", &format!("/sessions/{session_id}/interrupt"), None) + } + + pub fn stop_session(&self, session_id: &str) -> Result<(), ApiError> { + self.empty_request("POST", &format!("/sessions/{session_id}/stop"), None) + } + + pub fn start_session(&self, session_id: &str) -> Result<(), ApiError> { + self.empty_request("POST", &format!("/sessions/{session_id}/start"), None) + } + + pub fn rename_session(&self, session_id: &str, title: &str) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/title"), + Some(serde_json::json!({ "title": title })), + ) + } + + pub fn set_session_cwd(&self, session_id: &str, cwd: &str) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/cwd"), + Some(serde_json::json!({ "cwd": cwd })), + ) + } + + pub fn set_session_model(&self, session_id: &str, model: &str) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/model"), + Some(serde_json::json!({ "model": model })), + ) + } + + pub fn set_session_permission_mode( + &self, + session_id: &str, + mode: &str, + ) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/permission-mode"), + Some(serde_json::json!({ "permissionMode": mode })), + ) + } + + pub fn set_session_notify(&self, session_id: &str, notify: bool) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/notify"), + Some(serde_json::json!({ "notify": notify })), + ) + } + + pub fn run_command(&self, session_id: &str, text: &str) -> Result<(), ApiError> { + self.empty_request( + "POST", + &format!("/sessions/{session_id}/command"), + Some(serde_json::json!({ "text": text })), + ) + } + + pub fn compact_session(&self, session_id: &str) -> Result<(), ApiError> { + self.empty_request("POST", &format!("/sessions/{session_id}/compact"), None) + } + + pub fn delete_session(&self, session_id: &str, delete_foreign: bool) -> Result<(), ApiError> { + let path = if delete_foreign { + format!("/sessions/{session_id}?deleteForeign=true") + } else { + format!("/sessions/{session_id}") + }; + self.empty_request("DELETE", &path, None) + } + + /// A page of transcript history. `before` is the newest-first cursor + /// (server default is "the newest page" when absent, which a caller + /// gets by passing `None`); the events themselves are handed back as + /// [`event_model::SeqEvent`] via `crate::event_stream`'s parsing, kept + /// out of this method's signature so a caller that only wants the raw + /// lines (for the transcript cache) is not forced to parse them. + pub fn fetch_transcript_page( + &self, + session_id: &str, + before: Option, + limit: u32, + coalesce: bool, + ) -> Result, ApiError> { + let mut path = format!("/sessions/{session_id}/transcript?limit={limit}"); + if let Some(before) = before { + path.push_str(&format!("&before={before}")); + } + if coalesce { + path.push_str("&coalesce=true"); + } + self.json_request("GET", &path, None) + } +} + +/// The blocking [`Transport`] backed by `ureq`, the same crate `server/` +/// already depends on for its own outbound HTTPS (`usage.rs`'s Anthropic +/// poll). Verifies the server's leaf against a single pinned CA, the way +/// `ServerConfig.kt`'s `applyPinnedTls` does, rather than the system trust +/// store -- the server's certificate is self-signed on purpose (see +/// `wg-app-link`). +pub struct UreqTransport { + agent: ureq::Agent, + base_url: String, + token: String, +} + +impl UreqTransport { + /// `ca_pem` is the CA certificate `wg-app-link`'s `enroll` minted, + /// exactly as read from `certs/ca.pem`. + pub fn new( + base_url: impl Into, + token: impl Into, + ca_pem: &[u8], + ) -> Result { + let cert = ureq::tls::Certificate::from_pem(ca_pem).map_err(|e| ApiError { + message: format!("The pinned CA certificate could not be read: {e}"), + status: None, + })?; + let tls_config = ureq::tls::TlsConfig::builder() + .root_certs(ureq::tls::RootCerts::new_with_certs(&[cert])) + .build(); + let agent: ureq::Agent = ureq::Agent::config_builder() + .tls_config(tls_config) + // Read the body ourselves on every status, the way + // `requestFromServer` does: the server's own error wording is + // in the body of a 4xx/5xx, and the default behaviour throws + // it away before this code can read it. + .http_status_as_error(false) + .timeout_connect(Some(std::time::Duration::from_secs(5))) + .build() + .into(); + Ok(Self { + agent, + base_url: base_url.into(), + token: token.into(), + }) + } + + fn url(&self, path: &str) -> String { + format!("{}{}", self.base_url, path) + } +} + +impl Transport for UreqTransport { + fn request( + &self, + method: &str, + path: &str, + body: Option, + ) -> Result { + let url = self.url(path); + let auth = format!("Bearer {}", self.token); + let mut builder = ureq::http::Request::builder() + .method(method) + .uri(&url) + .header("Authorization", &auth); + let response = match body { + None => builder + .body(()) + .map_err(ureq::Error::from) + .and_then(|req| self.agent.run(req)), + Some(Body::Json(value)) => { + builder = builder.header("Content-Type", "application/json"); + builder + .body(serde_json::to_vec(&value).unwrap_or_default()) + .map_err(ureq::Error::from) + .and_then(|req| self.agent.run(req)) + } + Some(Body::Bytes { + content_type, + bytes, + }) => { + builder = builder.header("Content-Type", content_type); + builder + .body(bytes) + .map_err(ureq::Error::from) + .and_then(|req| self.agent.run(req)) + } + }; + let mut response = response.map_err(|e| transport_error(&self.base_url, path, e))?; + let status = response.status().as_u16(); + let mut body = Vec::new(); + response + .body_mut() + .as_reader() + .read_to_end(&mut body) + .map_err(|e| ApiError { + message: format!("Reached {url} but couldn't read its response ({e})"), + status: Some(status), + })?; + if !(200..300).contains(&status) { + return Err(response_error(status, &body, path)); + } + Ok(RawResponse { status, body }) + } + + fn stream(&self, path: &str) -> Result, ApiError> { + let url = self.url(path); + let auth = format!("Bearer {}", self.token); + let response = self + .agent + .get(&url) + .header("Authorization", &auth) + .header("Accept", "text/event-stream") + // No read timeout: between events there is nothing to read for + // as long as the thing being followed is idle, mirroring + // `EventStream.kt`'s `readTimeout = 0`. + .config() + .timeout_recv_response(None) + .build() + .call(); + let mut response = response.map_err(|e| transport_error(&self.base_url, path, e))?; + let status = response.status().as_u16(); + if status != 200 { + let mut body = Vec::new(); + let _ = response.body_mut().as_reader().read_to_end(&mut body); + return Err(response_error(status, &body, path)); + } + Ok(Box::new(response.into_body().into_reader())) + } +} + +fn transport_error(base_url: &str, path: &str, e: ureq::Error) -> ApiError { + ApiError { + message: format!( + "Couldn't reach the server at {base_url} ({e}) -- is ai-server running, and is this \ + device able to reach that address (WireGuard up)? [{path}]" + ), + status: None, + } +} + +/// The 401 wording matches `Api.kt`'s, since that message is instructions +/// for the reader rather than a diagnostic -- see this project's UI rule +/// about shortening a failure in one place rather than at each display site. +fn response_error(status: u16, body: &[u8], path: &str) -> ApiError { + let detail = String::from_utf8_lossy(body).trim().to_string(); + let message = if status == 401 { + "The server rejected this device's token. Re-enroll by scanning the server's QR (or \ + rotate with --rotate-token and scan the new one)." + .to_string() + } else if detail.is_empty() { + format!("Server returned HTTP {status} for {path}") + } else { + detail + }; + ApiError { + message, + status: Some(status), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Cursor; + use std::sync::Mutex; + + /// A transport with no network at all, for the pure-logic tests this + /// module can run without a server. + #[derive(Default)] + struct FakeTransport { + responses: Mutex>, + } + + impl FakeTransport { + fn respond(&self, method: &str, path: &str, status: u16, body: &str) { + self.responses.lock().unwrap().push(( + method.to_string(), + path.to_string(), + RawResponse { + status, + body: body.as_bytes().to_vec(), + }, + )); + } + } + + impl Transport for FakeTransport { + fn request( + &self, + method: &str, + path: &str, + _body: Option, + ) -> Result { + let mut responses = self.responses.lock().unwrap(); + let index = responses + .iter() + .position(|(m, p, _)| m == method && p == path) + .ok_or_else(|| ApiError { + message: format!("no fake response for {method} {path}"), + status: None, + })?; + let (_, _, response) = responses.remove(index); + if !(200..300).contains(&response.status) { + return Err(response_error(response.status, &response.body, path)); + } + Ok(response) + } + + fn stream(&self, _path: &str) -> Result, ApiError> { + Ok(Box::new(Cursor::new(Vec::new()))) + } + } + + #[test] + fn fetch_sessions_parses_the_list() { + let transport = FakeTransport::default(); + transport.respond( + "GET", + "/sessions", + 200, + r#"[{"id":"s1","setup":"m1","setupName":"desktop","provider":"claude_cli", + "title":"hi","status":"idle","lastActivity":1.0}]"#, + ); + let client = ApiClient::new(transport); + let sessions = client.fetch_sessions().unwrap(); + assert_eq!(sessions.len(), 1); + assert_eq!(sessions[0].id, "s1"); + assert_eq!(sessions[0].setup_name, "desktop"); + // Defaults for fields the server omits. + assert!(sessions[0].notify); + assert_eq!(sessions[0].model, None); + } + + #[test] + fn a_401_gets_the_enrollment_message_regardless_of_the_bare_body() { + let transport = FakeTransport::default(); + transport.respond("POST", "/sessions/s1/interrupt", 401, "unauthorized"); + let client = ApiClient::new(transport); + let err = client.interrupt_session("s1").unwrap_err(); + assert!(err.message.contains("Re-enroll")); + assert_eq!(err.status, Some(401)); + } + + #[test] + fn a_bare_error_status_with_no_body_falls_back_to_a_generic_message() { + let transport = FakeTransport::default(); + transport.respond("POST", "/sessions/s1/stop", 500, ""); + let client = ApiClient::new(transport); + let err = client.stop_session("s1").unwrap_err(); + assert!(err.message.contains("500")); + } + + #[test] + fn a_server_explanation_in_the_body_is_surfaced_verbatim() { + let transport = FakeTransport::default(); + transport.respond( + "POST", + "/sessions/s1/cwd", + 409, + "that path does not exist on this machine", + ); + let client = ApiClient::new(transport); + let err = client.set_session_cwd("s1", "/nope").unwrap_err(); + assert_eq!(err.message, "that path does not exist on this machine"); + } +} diff --git a/client-core/src/event_stream.rs b/client-core/src/event_stream.rs new file mode 100644 index 0000000..b170857 --- /dev/null +++ b/client-core/src/event_stream.rs @@ -0,0 +1,142 @@ +//! 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, + ) -> Result { + unimplemented!("this fixture only serves a stream") + } + + fn stream(&self, _path: &str) -> Result, 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); + } +} diff --git a/client-core/src/lib.rs b/client-core/src/lib.rs index db0bc34..28c9cc5 100644 --- a/client-core/src/lib.rs +++ b/client-core/src/lib.rs @@ -3,7 +3,10 @@ //! not yet. pub mod ansi; +pub mod api; +pub mod event_stream; pub mod highlight; +pub mod sse; pub mod transcript_cache; pub use event_model::*; diff --git a/client-core/src/sse.rs b/client-core/src/sse.rs new file mode 100644 index 0000000..87aef26 --- /dev/null +++ b/client-core/src/sse.rs @@ -0,0 +1,121 @@ +//! Server-sent-events framing, ported from `app/.../Sse.kt`: `data:` and +//! `event:` lines accumulate until a blank line ends the frame, comments +//! start with `:`, and a frame is either named with no payload or a payload +//! with no name. +//! +//! Pure and line-at-a-time, unlike the Kotlin original which also owned the +//! socket: `server/routes.rs`'s SSE bodies are one event per line, so a +//! caller here feeds lines from wherever they came from (a real connection, +//! a test fixture) and gets frames back with no I/O of its own -- which is +//! what lets this be tested with no server, per RUST.md's "pure logic +//! first" for this crate. + +/// One SSE frame: its name (`None` for an ordinary data frame) and its payload. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Frame { + pub name: Option, + pub data: String, +} + +/// Accumulates lines into [`Frame`]s. One instance per connection -- +/// `feed_line` is called for every line the transport reads (with line +/// endings already stripped), and answers a frame when a blank line closes +/// one. +#[derive(Debug, Default)] +pub struct SseReader { + data: String, + name: Option, +} + +impl SseReader { + pub fn new() -> Self { + Self::default() + } + + /// Feeds one line (no trailing `\n`). Answers the frame this line + /// completed, if any. + pub fn feed_line(&mut self, line: &str) -> Option { + if line.is_empty() { + if self.name.is_some() || !self.data.is_empty() { + let frame = Frame { + name: self.name.take(), + data: std::mem::take(&mut self.data), + }; + return Some(frame); + } + return None; + } + if let Some(rest) = line.strip_prefix("data:") { + self.data.push_str(rest.trim()); + } else if let Some(rest) = line.strip_prefix("event:") { + self.name = Some(rest.trim().to_string()); + } + // `id:`, comments -- nothing to do. + None + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn frames(lines: &[&str]) -> Vec { + let mut reader = SseReader::new(); + lines.iter().filter_map(|l| reader.feed_line(l)).collect() + } + + #[test] + fn a_data_only_frame_has_no_name() { + assert_eq!( + frames(&["data:hello", ""]), + vec![Frame { + name: None, + data: "hello".to_string() + }] + ); + } + + #[test] + fn a_named_frame_with_no_payload_still_completes() { + assert_eq!( + frames(&["event:reset", ""]), + vec![Frame { + name: Some("reset".to_string()), + data: String::new() + }] + ); + } + + #[test] + fn a_blank_line_with_nothing_pending_yields_no_frame() { + assert_eq!(frames(&[""]), vec![]); + } + + #[test] + fn a_comment_and_an_id_line_are_ignored() { + assert_eq!( + frames(&[":keepalive", "id:5", "data:hi", ""]), + vec![Frame { + name: None, + data: "hi".to_string() + }] + ); + } + + #[test] + fn two_frames_in_a_row_are_both_reported() { + assert_eq!( + frames(&["data:one", "", "data:two", ""]), + vec![ + Frame { + name: None, + data: "one".to_string() + }, + Frame { + name: None, + data: "two".to_string() + }, + ] + ); + } +}