AkurAI-Framework / crates/ui/src/http.rs

Latest change 2ebaebff3945c75127afb6a9524ac1192993158b - akurai-ui: EventSource bridge (http::event_source, stream callbacks, host.js + native fake, testing helpers) by Ólafur Búi Ólafsson

//! `fetch` with JSON helpers, and `EventSource` streams. Same-origin, cookies
//! included.

use crate::callbacks;
use crate::host;
use akurai_json::Value;

/// A completed HTTP exchange. `status` is 0 on a network failure.
#[derive(Clone, Debug, PartialEq)]
pub struct Response {
    pub status: u16,
    pub body: String,
}

impl Response {
    pub fn ok(&self) -> bool {
        (200..300).contains(&self.status)
    }
    /// Parse the body as JSON (`Value::Null` when empty or invalid).
    pub fn json(&self) -> Value {
        akurai_json::parse(&self.body).unwrap_or(Value::Null)
    }
    /// The `error` field of a JSON error body, or a generic message.
    pub fn error_message(&self) -> String {
        match self.json().get("error").and_then(Value::as_str) {
            Some(msg) => msg.to_string(),
            None if self.status == 0 => "Network error — check your connection".to_string(),
            None => format!("Request failed ({})", self.status),
        }
    }
}

/// Send a request; `done` runs when it completes.
pub fn fetch(method: &str, url: &str, body: Option<&str>, done: impl FnOnce(Response) + 'static) {
    let mut done = Some(done);
    let mut run = crate::dom::in_scope(crate::reactive::current_scope(), move |r: Response| {
        if let Some(done) = done.take() {
            done(r)
        }
    });
    let id = callbacks::once(move |status, body| run(Response { status, body }));
    host::fetch(method, url, body, id);
}

pub fn get(url: &str, done: impl FnOnce(Response) + 'static) {
    fetch("GET", url, None, done)
}

/// Send `body` as JSON.
pub fn send_json(method: &str, url: &str, body: &Value, done: impl FnOnce(Response) + 'static) {
    fetch(method, url, Some(&body.to_json()), done)
}

/// One message from an [`event_source`] stream.
#[derive(Clone, Debug, PartialEq)]
pub enum StreamEvent {
    /// A server event: its name (`message` when unnamed) and data.
    Message { name: String, data: String },
    /// The connection dropped; the browser is reconnecting (`closed` when it
    /// gave up). Nothing to do but show it.
    Error(String),
}

/// An open stream; dropping it does nothing, [`EventSource::close`] ends it.
/// A stream opened inside a component scope closes when that scope ends.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct EventSource {
    id: u32,
    callback: u32,
}

impl EventSource {
    pub fn close(self) {
        host::event_source_close(self.id);
        callbacks::remove(self.callback);
    }
}

/// Subscribe to a Server-Sent Events stream. `names` are the named events to
/// listen for (unnamed ones always arrive as `message`). `on` runs per event,
/// in the scope that opened the stream, and the stream closes with that scope.
pub fn event_source(
    url: &str,
    names: &[&str],
    on: impl FnMut(StreamEvent) + 'static,
) -> EventSource {
    let mut run = crate::dom::in_scope(crate::reactive::current_scope(), on);
    let callback = callbacks::stream(move |status, body| {
        let event = if status == 1 {
            let (name, data) = body.split_once('\n').unwrap_or((&body, ""));
            StreamEvent::Message {
                name: name.to_string(),
                data: data.to_string(),
            }
        } else {
            StreamEvent::Error(body)
        };
        run(event)
    });
    let id = host::event_source(url, &names.join(","), callback);
    let stream = EventSource { id, callback };
    crate::reactive::on_cleanup(move || stream.close());
    stream
}

#[cfg(test)]
mod stream_tests {
    use super::*;
    use crate::reactive::{create_root, signal};
    use crate::testing as t;

    #[test]
    fn an_event_source_delivers_named_events_and_errors_and_closes_with_its_scope() {
        t::reset();
        let seen = signal(Vec::<String>::new());
        let live = callbacks::live();
        let (_, root) = create_root(move || {
            event_source("/api/home/events", &["snapshot", "changed"], move |e| {
                seen.update(|v| v.push(format!("{e:?}")))
            });
        });
        let streams = t::open_streams();
        assert_eq!(streams.len(), 1);
        assert_eq!(streams[0].url, "/api/home/events");
        assert_eq!(streams[0].names, vec!["snapshot", "changed"]);

        t::push_event("/api/home/events", "changed", "{\"id\":\"light.a\"}");
        t::push_event("/api/home/events", "message", "hi");
        t::stream_error("/api/home/events", "reconnecting");
        let got = seen.get();
        assert_eq!(got.len(), 3, "{got:?}");
        assert!(got[0].contains("name: \"changed\"") && got[0].contains("light.a"));
        assert!(got[1].contains("name: \"message\""));
        assert_eq!(got[2], "Error(\"reconnecting\")");

        root.dispose();
        assert!(t::open_streams().is_empty(), "closed with its scope");
        assert_eq!(callbacks::live(), live, "no leaked callback");
    }
}