nordisk/edda

crates/nornir-saga/src/ui/source.rs

//! **The source seam** — where a viewer/replay pulls its [`TimedUiMsg`] stream
//! from. Two built-in sources:
//!
//! * [`RecordedSource`] — a recorded [`Saga`] / `Vec` (the demo + replay path);
//! * [`ChannelSource`] — a live stream over an `mpsc` channel (the LIVE-view
//!   path a saga-enabled Njord's SSE/socket feed lands in).
//!
//! Both implement [`SagaSource`], so the LIVE viewer and the recorded replay
//! consume movements through the same trait. (The web frontend fans a
//! [`Saga`] out over SSE using the recorded source; a live Njord feed would use
//! a [`ChannelSource`] fed by the socket task.)

use std::sync::mpsc::Receiver;

use crate::ui::msg::{Saga, TimedUiMsg};

/// A pull-based source of timed UI movements. `next` returns the next movement
/// in order, or `None` when the stream is exhausted (recorded) / closed (live).
pub trait SagaSource {
    /// The next movement, or `None` at end-of-stream.
    fn next(&mut self) -> Option<TimedUiMsg>;
}

/// A source backed by a recorded [`Saga`] (a `Vec<TimedUiMsg>`). Iterates the
/// events in recorded order. This is the deterministic demo + replay source.
#[derive(Debug, Clone)]
pub struct RecordedSource {
    events: Vec<TimedUiMsg>,
    pos: usize,
}

impl RecordedSource {
    /// A source over a whole saga.
    pub fn new(saga: Saga) -> RecordedSource {
        RecordedSource { events: saga.events, pos: 0 }
    }

    /// A source over a raw `Vec` of movements.
    pub fn from_events(events: Vec<TimedUiMsg>) -> RecordedSource {
        RecordedSource { events, pos: 0 }
    }

    /// Rewind to the start so the saga can be replayed again.
    pub fn rewind(&mut self) {
        self.pos = 0;
    }

    /// Movements remaining.
    pub fn remaining(&self) -> usize {
        self.events.len().saturating_sub(self.pos)
    }
}

impl SagaSource for RecordedSource {
    fn next(&mut self) -> Option<TimedUiMsg> {
        let ev = self.events.get(self.pos).cloned();
        if ev.is_some() {
            self.pos += 1;
        }
        ev
    }
}

/// A source backed by a live channel — the LIVE-view path. A socket/SSE task
/// feeding a saga-enabled Njord's `Msg` stream sends [`TimedUiMsg`]s down the
/// channel; the viewer pulls them here. `next` **blocks** until the next
/// movement arrives or the sender is dropped (stream closed → `None`).
pub struct ChannelSource {
    rx: Receiver<TimedUiMsg>,
}

impl ChannelSource {
    /// Wrap a receiver whose sender the live feed task holds.
    pub fn new(rx: Receiver<TimedUiMsg>) -> ChannelSource {
        ChannelSource { rx }
    }
}

impl SagaSource for ChannelSource {
    fn next(&mut self) -> Option<TimedUiMsg> {
        // Ok → a movement; Err(RecvError) → all senders dropped, stream closed.
        self.rx.recv().ok()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ui::msg::UiMsg;

    #[test]
    fn recorded_source_yields_in_order_then_none() {
        let saga = Saga::new("t")
            .push(UiMsg::nav("/").at(0))
            .push(UiMsg::pointer_move(1.0, 2.0).at(10));
        let mut src = RecordedSource::new(saga);
        assert_eq!(src.remaining(), 2);
        assert!(matches!(src.next().unwrap().msg, UiMsg::Nav { .. }));
        assert!(matches!(src.next().unwrap().msg, UiMsg::PointerMove { .. }));
        assert!(src.next().is_none(), "exhausted → None");
        src.rewind();
        assert_eq!(src.remaining(), 2, "rewind restores the stream");
    }

    #[test]
    fn channel_source_streams_then_closes() {
        let (tx, rx) = std::sync::mpsc::channel();
        tx.send(UiMsg::nav("/live").at(0)).unwrap();
        tx.send(UiMsg::click(5.0, 5.0, "#x").at(50)).unwrap();
        drop(tx); // close the stream
        let mut src = ChannelSource::new(rx);
        assert!(matches!(src.next().unwrap().msg, UiMsg::Nav { .. }));
        assert!(matches!(src.next().unwrap().msg, UiMsg::Click { .. }));
        assert!(src.next().is_none(), "closed sender → end of stream");
    }
}