koi_client/
lib.rs

1//! HTTP client for communicating with a running Koi daemon.
2//!
3//! Uses blocking `ureq` - no async runtime dependency on the client path.
4//! All paths use `/v1/mdns/` prefix for mDNS domain routes.
5
6use std::io::{BufRead, BufReader, Read};
7use std::time::Duration;
8
9use hickory_proto::rr::RecordType;
10use koi_common::mdns_protocol::{
11    AdminRegistration, DaemonStatus, RegisterPayload, RegistrationResult, RenewalResult,
12};
13use koi_common::net::resolve_localhost;
14use koi_common::types::{ServiceCheckKind, ServiceRecord};
15
16/// TCP connection timeout for general API requests.
17const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
18
19/// Read timeout for general (non-streaming) API requests.
20const READ_TIMEOUT: Duration = Duration::from_secs(10);
21
22/// Timeout for the fast health check probe.
23const HEALTH_TIMEOUT: Duration = Duration::from_millis(200);
24
25// ── Error types ───────────────────────────────────────────────────
26
27#[derive(Debug, thiserror::Error)]
28pub enum ClientError {
29    #[error("Daemon not reachable: {0}")]
30    Unreachable(String),
31
32    /// HTTP 401 from the daemon: the request needs a Daemon Access Token.
33    /// Surfaced distinctly (instead of a generic `Api`) so the CLI can print an
34    /// actionable hint when talking to an explicit `--endpoint`.
35    #[error("remote daemon requires a token (pass --token or set KOI_TOKEN)")]
36    Unauthorized,
37
38    #[error("{error}: {message}")]
39    Api { error: String, message: String },
40
41    #[error("Request failed: {0}")]
42    Transport(String),
43
44    #[error("Invalid response: {0}")]
45    Decode(String),
46}
47
48impl ClientError {
49    /// Whether this error is an HTTP 401 (missing/invalid token).
50    pub fn is_unauthorized(&self) -> bool {
51        matches!(self, ClientError::Unauthorized)
52    }
53}
54
55pub type Result<T> = std::result::Result<T, ClientError>;
56
57// ── Client ────────────────────────────────────────────────────────
58
59/// Header name for Daemon Access Token authentication.
60const DAT_HEADER: &str = "X-Koi-Token";
61
62pub struct KoiClient {
63    endpoint: String,
64    agent: ureq::Agent,
65    /// Daemon Access Token (empty string means no auth).
66    token: String,
67}
68
69impl KoiClient {
70    pub fn new(endpoint: &str) -> Self {
71        let clean = endpoint.trim_end_matches('/');
72        let resolved = resolve_localhost(clean);
73        let agent = ureq::AgentBuilder::new()
74            .timeout_connect(CONNECT_TIMEOUT)
75            .timeout_read(READ_TIMEOUT)
76            .build();
77        Self {
78            endpoint: resolved,
79            agent,
80            token: String::new(),
81        }
82    }
83
84    /// Create a client with a Daemon Access Token for authenticated requests.
85    pub fn with_token(endpoint: &str, token: &str) -> Self {
86        let mut client = Self::new(endpoint);
87        client.token = token.to_string();
88        client
89    }
90
91    /// Create a client from the breadcrumb file (endpoint + token).
92    ///
93    /// Returns `None` if no breadcrumb exists.
94    pub fn from_breadcrumb() -> Option<Self> {
95        let bc = koi_config::breadcrumb::read_breadcrumb()?;
96        Some(Self::with_token(&bc.endpoint, &bc.token))
97    }
98
99    /// Attach the DAT header to a request if a token is present.
100    fn auth_get(&self, url: &str) -> ureq::Request {
101        let req = self.agent.get(url);
102        if self.token.is_empty() {
103            req
104        } else {
105            req.set(DAT_HEADER, &self.token)
106        }
107    }
108
109    /// Attach the DAT header to a POST request.
110    fn auth_post(&self, url: &str) -> ureq::Request {
111        let req = self.agent.post(url);
112        if self.token.is_empty() {
113            req
114        } else {
115            req.set(DAT_HEADER, &self.token)
116        }
117    }
118
119    /// Attach the DAT header to a PUT request.
120    fn auth_put(&self, url: &str) -> ureq::Request {
121        let req = self.agent.put(url);
122        if self.token.is_empty() {
123            req
124        } else {
125            req.set(DAT_HEADER, &self.token)
126        }
127    }
128
129    /// Attach the DAT header to a DELETE request.
130    fn auth_delete(&self, url: &str) -> ureq::Request {
131        let req = self.agent.delete(url);
132        if self.token.is_empty() {
133            req
134        } else {
135            req.set(DAT_HEADER, &self.token)
136        }
137    }
138
139    // ── Health ────────────────────────────────────────────────────
140
141    /// Quick health check with a 200ms timeout.
142    pub fn health(&self) -> Result<()> {
143        let agent = ureq::AgentBuilder::new()
144            .timeout_connect(HEALTH_TIMEOUT)
145            .timeout_read(HEALTH_TIMEOUT)
146            .build();
147        let url = format!("{}/healthz", self.endpoint);
148        agent.get(&url).call().map_err(map_error)?;
149        Ok(())
150    }
151
152    // ── Service operations (mDNS) ──────────────────────────────────
153
154    pub fn register(&self, payload: &RegisterPayload) -> Result<RegistrationResult> {
155        let url = format!("{}/v1/mdns/announce", self.endpoint);
156        let json_val =
157            serde_json::to_value(payload).map_err(|e| ClientError::Decode(e.to_string()))?;
158        let resp = self
159            .auth_post(&url)
160            .send_json(json_val)
161            .map_err(map_error)?;
162        let json: serde_json::Value = resp
163            .into_json()
164            .map_err(|e| ClientError::Decode(e.to_string()))?;
165        extract(&json, "registered")
166    }
167
168    pub fn unregister(&self, id: &str) -> Result<()> {
169        let url = format!("{}/v1/mdns/unregister/{id}", self.endpoint);
170        self.auth_delete(&url).call().map_err(map_error)?;
171        Ok(())
172    }
173
174    pub fn heartbeat(&self, id: &str) -> Result<RenewalResult> {
175        let url = format!("{}/v1/mdns/heartbeat/{id}", self.endpoint);
176        let resp = self.auth_put(&url).send_bytes(&[]).map_err(map_error)?;
177        let json: serde_json::Value = resp
178            .into_json()
179            .map_err(|e| ClientError::Decode(e.to_string()))?;
180        extract(&json, "renewed")
181    }
182
183    pub fn resolve(&self, instance: &str) -> Result<ServiceRecord> {
184        let url = format!("{}/v1/mdns/resolve", self.endpoint);
185        let resp = self
186            .auth_get(&url)
187            .query("name", instance)
188            .call()
189            .map_err(map_error)?;
190        let json: serde_json::Value = resp
191            .into_json()
192            .map_err(|e| ClientError::Decode(e.to_string()))?;
193        extract(&json, "resolved")
194    }
195
196    /// Start a browse SSE stream. Returns an iterator of JSON events.
197    pub fn browse_stream(&self, service_type: &str) -> Result<SseStream> {
198        let url = format!("{}/v1/mdns/discover", self.endpoint);
199        let mut req = self.stream_agent().get(&url);
200        if !self.token.is_empty() {
201            req = req.set(DAT_HEADER, &self.token);
202        }
203        let resp = req.query("type", service_type).call().map_err(map_error)?;
204        Ok(SseStream::new(Box::new(resp.into_reader())))
205    }
206
207    /// Start an events SSE stream. Returns an iterator of JSON events.
208    pub fn events_stream(&self, service_type: &str) -> Result<SseStream> {
209        let url = format!("{}/v1/mdns/subscribe", self.endpoint);
210        let mut req = self.stream_agent().get(&url);
211        if !self.token.is_empty() {
212            req = req.set(DAT_HEADER, &self.token);
213        }
214        let resp = req.query("type", service_type).call().map_err(map_error)?;
215        Ok(SseStream::new(Box::new(resp.into_reader())))
216    }
217
218    // ── Unified status ─────────────────────────────────────────────
219
220    /// Fetch unified status from `/v1/status`.
221    pub fn unified_status(&self) -> Result<serde_json::Value> {
222        let url = format!("{}/v1/status", self.endpoint);
223        let resp = self.auth_get(&url).call().map_err(map_error)?;
224        resp.into_json()
225            .map_err(|e| ClientError::Decode(e.to_string()))
226    }
227
228    // ── DNS operations (Phase 6) ───────────────────────────────────
229
230    pub fn dns_status(&self) -> Result<serde_json::Value> {
231        self.get_json("/v1/dns/status")
232    }
233
234    pub fn dns_lookup(&self, name: &str, record_type: RecordType) -> Result<serde_json::Value> {
235        let url = format!("{}/v1/dns/lookup", self.endpoint);
236        let resp = self
237            .auth_get(&url)
238            .query("name", name)
239            .query("type", record_type_str(record_type))
240            .call()
241            .map_err(map_error)?;
242        resp.into_json()
243            .map_err(|e| ClientError::Decode(e.to_string()))
244    }
245
246    pub fn dns_list(&self) -> Result<serde_json::Value> {
247        self.get_json("/v1/dns/list")
248    }
249
250    pub fn dns_add(&self, name: &str, ip: &str, ttl: Option<u32>) -> Result<serde_json::Value> {
251        let body = serde_json::json!({
252            "name": name,
253            "ip": ip,
254            "ttl": ttl,
255        });
256        self.post_json("/v1/dns/add", &body)
257    }
258
259    pub fn dns_remove(&self, name: &str) -> Result<serde_json::Value> {
260        let url = format!("{}/v1/dns/remove/{}", self.endpoint, name);
261        let resp = self.auth_delete(&url).call().map_err(map_error)?;
262        resp.into_json()
263            .map_err(|e| ClientError::Decode(e.to_string()))
264    }
265
266    pub fn dns_start(&self) -> Result<serde_json::Value> {
267        self.post_json("/v1/dns/serve", &serde_json::json!({}))
268    }
269
270    pub fn dns_stop(&self) -> Result<serde_json::Value> {
271        self.post_json("/v1/dns/stop", &serde_json::json!({}))
272    }
273
274    // ── Health operations (Phase 7) ───────────────────────────────
275
276    pub fn health_status(&self) -> Result<serde_json::Value> {
277        self.get_json("/v1/health/status")
278    }
279
280    pub fn health_add_check(
281        &self,
282        name: &str,
283        kind: ServiceCheckKind,
284        target: &str,
285        interval_secs: u64,
286        timeout_secs: u64,
287    ) -> Result<serde_json::Value> {
288        let body = serde_json::json!({
289            "name": name,
290            "kind": check_kind_str(kind),
291            "target": target,
292            "interval_secs": interval_secs,
293            "timeout_secs": timeout_secs,
294        });
295        self.post_json("/v1/health/add", &body)
296    }
297
298    pub fn health_remove_check(&self, name: &str) -> Result<serde_json::Value> {
299        let url = format!("{}/v1/health/remove/{}", self.endpoint, name);
300        let resp = self.auth_delete(&url).call().map_err(map_error)?;
301        resp.into_json()
302            .map_err(|e| ClientError::Decode(e.to_string()))
303    }
304
305    // ── Proxy operations (Phase 8) ───────────────────────────────
306
307    pub fn proxy_status(&self) -> Result<serde_json::Value> {
308        self.get_json("/v1/proxy/status")
309    }
310
311    pub fn proxy_list(&self) -> Result<serde_json::Value> {
312        self.get_json("/v1/proxy/list")
313    }
314
315    pub fn proxy_add(
316        &self,
317        name: &str,
318        listen_port: u16,
319        backend: &str,
320        allow_remote: bool,
321    ) -> Result<serde_json::Value> {
322        let body = serde_json::json!({
323            "name": name,
324            "listen_port": listen_port,
325            "backend": backend,
326            "allow_remote": allow_remote,
327        });
328        self.post_json("/v1/proxy/add", &body)
329    }
330
331    pub fn proxy_remove(&self, name: &str) -> Result<serde_json::Value> {
332        let url = format!("{}/v1/proxy/remove/{}", self.endpoint, name);
333        let resp = self.auth_delete(&url).call().map_err(map_error)?;
334        resp.into_json()
335            .map_err(|e| ClientError::Decode(e.to_string()))
336    }
337
338    // ── UDP operations ─────────────────────────────────────────────
339
340    pub fn udp_status(&self) -> Result<serde_json::Value> {
341        self.get_json("/v1/udp/status")
342    }
343
344    pub fn udp_bind(&self, port: u16, addr: &str, lease_secs: u64) -> Result<serde_json::Value> {
345        let body = serde_json::json!({
346            "port": port,
347            "addr": addr,
348            "lease_secs": lease_secs,
349        });
350        self.post_json("/v1/udp/bind", &body)
351    }
352
353    pub fn udp_unbind(&self, id: &str) -> Result<serde_json::Value> {
354        let url = format!("{}/v1/udp/bind/{}", self.endpoint, id);
355        let resp = self.auth_delete(&url).call().map_err(map_error)?;
356        resp.into_json()
357            .map_err(|e| ClientError::Decode(e.to_string()))
358    }
359
360    pub fn udp_send(&self, id: &str, dest: &str, payload_b64: &str) -> Result<serde_json::Value> {
361        let body = serde_json::json!({
362            "dest": dest,
363            "payload": payload_b64,
364        });
365        let path = format!("/v1/udp/send/{id}");
366        self.post_json(&path, &body)
367    }
368
369    pub fn udp_heartbeat(&self, id: &str) -> Result<serde_json::Value> {
370        let path = format!("/v1/udp/heartbeat/{id}");
371        self.put_json(&path, &serde_json::json!({}))
372    }
373
374    // ── Generic operations ─────────────────────────────────────────
375
376    /// POST JSON to an arbitrary path and return the response as a JSON value.
377    pub fn post_json(&self, path: &str, body: &serde_json::Value) -> Result<serde_json::Value> {
378        let url = format!("{}{path}", self.endpoint);
379        let resp = self
380            .auth_post(&url)
381            .send_json(body.clone())
382            .map_err(map_error)?;
383        resp.into_json()
384            .map_err(|e| ClientError::Decode(e.to_string()))
385    }
386
387    /// GET JSON from an arbitrary path and return the response as a JSON value.
388    pub fn get_json(&self, path: &str) -> Result<serde_json::Value> {
389        let url = format!("{}{path}", self.endpoint);
390        let resp = self.auth_get(&url).call().map_err(map_error)?;
391        resp.into_json()
392            .map_err(|e| ClientError::Decode(e.to_string()))
393    }
394
395    /// PUT JSON to an arbitrary path and return the response as a JSON value.
396    pub fn put_json(&self, path: &str, body: &serde_json::Value) -> Result<serde_json::Value> {
397        let url = format!("{}{path}", self.endpoint);
398        let resp = self
399            .auth_put(&url)
400            .send_json(body.clone())
401            .map_err(map_error)?;
402        resp.into_json()
403            .map_err(|e| ClientError::Decode(e.to_string()))
404    }
405
406    // ── Admin operations (mDNS) ──────────────────────────────────
407
408    pub fn admin_status(&self) -> Result<DaemonStatus> {
409        let url = format!("{}/v1/mdns/admin/status", self.endpoint);
410        let resp = self.auth_get(&url).call().map_err(map_error)?;
411        resp.into_json()
412            .map_err(|e| ClientError::Decode(e.to_string()))
413    }
414
415    pub fn admin_registrations(&self) -> Result<Vec<AdminRegistration>> {
416        let url = format!("{}/v1/mdns/admin/ls", self.endpoint);
417        let resp = self.auth_get(&url).call().map_err(map_error)?;
418        resp.into_json()
419            .map_err(|e| ClientError::Decode(e.to_string()))
420    }
421
422    pub fn admin_inspect(&self, id: &str) -> Result<AdminRegistration> {
423        let url = format!("{}/v1/mdns/admin/inspect/{id}", self.endpoint);
424        let resp = self.auth_get(&url).call().map_err(map_error)?;
425        resp.into_json()
426            .map_err(|e| ClientError::Decode(e.to_string()))
427    }
428
429    pub fn admin_force_unregister(&self, id: &str) -> Result<()> {
430        let url = format!("{}/v1/mdns/admin/unregister/{id}", self.endpoint);
431        self.auth_delete(&url).call().map_err(map_error)?;
432        Ok(())
433    }
434
435    pub fn admin_drain(&self, id: &str) -> Result<()> {
436        let url = format!("{}/v1/mdns/admin/drain/{id}", self.endpoint);
437        self.auth_post(&url).call().map_err(map_error)?;
438        Ok(())
439    }
440
441    pub fn admin_revive(&self, id: &str) -> Result<()> {
442        let url = format!("{}/v1/mdns/admin/revive/{id}", self.endpoint);
443        self.auth_post(&url).call().map_err(map_error)?;
444        Ok(())
445    }
446
447    // ── Admin operations (system) ────────────────────────────────────
448
449    /// Request a graceful shutdown of the running daemon.
450    pub fn shutdown(&self) -> Result<()> {
451        let url = format!("{}/v1/admin/shutdown", self.endpoint);
452        self.auth_post(&url).call().map_err(map_error)?;
453        Ok(())
454    }
455
456    // ── Private helpers ───────────────────────────────────────────
457
458    /// Agent without read timeout for SSE streams.
459    fn stream_agent(&self) -> ureq::Agent {
460        ureq::AgentBuilder::new()
461            .timeout_connect(CONNECT_TIMEOUT)
462            .build()
463    }
464}
465
466// ── SSE Stream ────────────────────────────────────────────────────
467
468/// Iterator over Server-Sent Events from the Koi daemon.
469///
470/// Parses `data: <json>` lines, skipping empty lines and event metadata.
471pub struct SseStream {
472    reader: BufReader<Box<dyn Read + Send>>,
473}
474
475impl SseStream {
476    fn new(reader: Box<dyn Read + Send>) -> Self {
477        Self {
478            reader: BufReader::new(reader),
479        }
480    }
481}
482
483impl Iterator for SseStream {
484    type Item = Result<serde_json::Value>;
485
486    fn next(&mut self) -> Option<Self::Item> {
487        loop {
488            let mut line = String::new();
489            match self.reader.read_line(&mut line) {
490                Ok(0) => return None,
491                Ok(_) => {
492                    let trimmed = line.trim();
493                    if let Some(data) = trimmed.strip_prefix("data:") {
494                        let data = data.trim_start();
495                        if data.is_empty() {
496                            continue;
497                        }
498                        match serde_json::from_str(data) {
499                            Ok(json) => return Some(Ok(json)),
500                            Err(e) => return Some(Err(ClientError::Decode(e.to_string()))),
501                        }
502                    }
503                    continue;
504                }
505                Err(e) => return Some(Err(ClientError::Transport(e.to_string()))),
506            }
507        }
508    }
509}
510
511// ── Error helpers ─────────────────────────────────────────────────
512
513fn map_error(e: ureq::Error) -> ClientError {
514    match e {
515        ureq::Error::Status(401, _resp) => ClientError::Unauthorized,
516        ureq::Error::Status(_status, resp) => {
517            let body = resp.into_string().unwrap_or_default();
518            if let Ok(json) = serde_json::from_str::<serde_json::Value>(&body) {
519                let error = json
520                    .get("error")
521                    .and_then(|v| v.as_str())
522                    .unwrap_or("unknown")
523                    .to_string();
524                let message = json
525                    .get("message")
526                    .and_then(|v| v.as_str())
527                    .unwrap_or(&body)
528                    .to_string();
529                ClientError::Api { error, message }
530            } else {
531                ClientError::Api {
532                    error: "http_error".into(),
533                    message: body,
534                }
535            }
536        }
537        ureq::Error::Transport(t) => ClientError::Unreachable(t.to_string()),
538    }
539}
540
541fn record_type_str(record_type: RecordType) -> &'static str {
542    match record_type {
543        RecordType::A => "A",
544        RecordType::AAAA => "AAAA",
545        RecordType::ANY => "ANY",
546        _ => "A",
547    }
548}
549
550fn check_kind_str(kind: ServiceCheckKind) -> &'static str {
551    match kind {
552        ServiceCheckKind::Http => "http",
553        ServiceCheckKind::Tcp => "tcp",
554    }
555}
556
557fn extract<T: serde::de::DeserializeOwned>(json: &serde_json::Value, key: &str) -> Result<T> {
558    if let Some(err_val) = json.get("error") {
559        let error = err_val.as_str().unwrap_or("unknown").to_string();
560        let message = json
561            .get("message")
562            .and_then(|m| m.as_str())
563            .unwrap_or("Unknown error")
564            .to_string();
565        return Err(ClientError::Api { error, message });
566    }
567    json.get(key)
568        .ok_or_else(|| ClientError::Decode(format!("Missing '{key}' in response")))
569        .and_then(|v| {
570            serde_json::from_value(v.clone()).map_err(|e| ClientError::Decode(e.to_string()))
571        })
572}
573
574#[cfg(test)]
575mod tests {
576    use super::*;
577
578    // ── Test helpers ────────────────────────────────────────────────
579
580    fn cursor_stream(input: &str) -> SseStream {
581        let cursor = std::io::Cursor::new(input.as_bytes().to_vec());
582        SseStream::new(Box::new(cursor))
583    }
584
585    // ── Unauthorized (401) hint tests ───────────────────────────────
586
587    #[test]
588    fn unauthorized_displays_actionable_hint() {
589        let err = ClientError::Unauthorized;
590        assert_eq!(
591            err.to_string(),
592            "remote daemon requires a token (pass --token or set KOI_TOKEN)"
593        );
594        assert!(err.is_unauthorized());
595    }
596
597    #[test]
598    fn non_401_api_error_is_not_unauthorized() {
599        let err = ClientError::Api {
600            error: "not_found".into(),
601            message: "nope".into(),
602        };
603        assert!(!err.is_unauthorized());
604    }
605
606    // ── KoiClient::new() tests ──────────────────────────────────────
607
608    #[test]
609    fn client_new_strips_trailing_slash() {
610        // After Happy Eyeballs, localhost is rewritten to a literal IP.
611        let client = KoiClient::new("http://localhost:5641/");
612        assert!(
613            client.endpoint == "http://127.0.0.1:5641"
614                || client.endpoint == "http://[::1]:5641"
615                || client.endpoint == "http://localhost:5641",
616            "unexpected endpoint: {}",
617            client.endpoint
618        );
619        assert!(!client.endpoint.ends_with("/"));
620        assert!(client.token.is_empty());
621    }
622
623    #[test]
624    fn client_with_token_sets_token() {
625        let client = KoiClient::with_token("http://10.0.0.1:5641", "my-secret-token");
626        assert_eq!(client.endpoint, "http://10.0.0.1:5641");
627        assert_eq!(client.token, "my-secret-token");
628    }
629
630    #[test]
631    fn client_new_preserves_non_localhost() {
632        let client = KoiClient::new("http://10.0.0.1:5641");
633        assert_eq!(client.endpoint, "http://10.0.0.1:5641");
634    }
635
636    #[test]
637    fn client_new_strips_multiple_trailing_slashes() {
638        let client = KoiClient::new("http://localhost:5641///");
639        assert!(!client.endpoint.ends_with("/"));
640    }
641
642    // ── SSE parsing tests ───────────────────────────────────────────
643
644    #[test]
645    fn sse_stream_yields_parsed_json() {
646        let input = "data: {\"foo\": 1}\n\n";
647        let mut stream = cursor_stream(input);
648        let item = stream.next().unwrap().unwrap();
649        assert_eq!(item["foo"], 1);
650    }
651
652    #[test]
653    fn sse_stream_skips_empty_lines() {
654        let input = "\n\n\n\n";
655        let mut stream = cursor_stream(input);
656        assert!(stream.next().is_none());
657    }
658
659    #[test]
660    fn sse_stream_skips_non_data_lines() {
661        let input = "event: message\nretry: 1000\n\n";
662        let mut stream = cursor_stream(input);
663        assert!(stream.next().is_none());
664    }
665
666    #[test]
667    fn sse_stream_handles_leading_space() {
668        let input = "data:   {\"hello\": \"world\"}\n";
669        let mut stream = cursor_stream(input);
670        let item = stream.next().unwrap().unwrap();
671        assert_eq!(item["hello"], "world");
672    }
673
674    #[test]
675    fn sse_stream_handles_no_space() {
676        let input = "data:{\"hello\":\"world\"}\n";
677        let mut stream = cursor_stream(input);
678        let item = stream.next().unwrap().unwrap();
679        assert_eq!(item["hello"], "world");
680    }
681
682    #[test]
683    fn sse_stream_yields_multiple_events() {
684        let input = "data: {\"n\": 1}\n\ndata: {\"n\": 2}\n\n";
685        let mut stream = cursor_stream(input);
686        let first = stream.next().unwrap().unwrap();
687        let second = stream.next().unwrap().unwrap();
688        assert_eq!(first["n"], 1);
689        assert_eq!(second["n"], 2);
690    }
691
692    #[test]
693    fn sse_stream_returns_none_on_eof() {
694        let input = "data: {\"n\": 1}\n";
695        let mut stream = cursor_stream(input);
696        let _ = stream.next();
697        assert!(stream.next().is_none());
698    }
699
700    #[test]
701    fn sse_stream_decode_error_on_invalid_json() {
702        let input = "data: {bad json}\n";
703        let mut stream = cursor_stream(input);
704        let item = stream.next().unwrap();
705        assert!(item.is_err());
706    }
707
708    #[test]
709    fn sse_stream_transport_error_on_read_failure() {
710        struct BrokenReader;
711        impl Read for BrokenReader {
712            fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
713                Err(std::io::Error::other("boom"))
714            }
715        }
716
717        let stream = SseStream::new(Box::new(BrokenReader));
718        let mut stream = stream;
719        let item = stream.next().unwrap();
720        assert!(item.is_err());
721    }
722}
koi_client/lib.rs

koi_client/
lib.rs
