crates/nornir-saga/src/ui/emit.rs
//! **The emit seam** — how a saga-enabled host *feeds* its Elm `Msg` stream to
//! nornir-saga. This is the explicit, testable counterpart to the zero-config
//! process-global [`tap`](crate::ui::tap):
//!
//! * The **global [`tap`]** is the one-liner an app drops at its Elm `update`
//! choke point; it buffers into a process-global saga and is compiled out of a
//! release build (feature `saga`). Zero config, one call site.
//! * The **[`Emitter`] + [`UiSink`]** here are the *pluggable* version: a host
//! owns an `Emitter<S>`, taps its `Msg`s through it, and the movements flow to
//! whatever sink it chose — an in-process [`VecSink`] (persist later), a live
//! [`ChannelSink`] (stream to a viewer thread — the LIVE path, paired with a
//! [`ChannelSource`](crate::ui::ChannelSource)), or [`NullSink`] (drop).
//!
//! Both project a host `Msg` to a [`UiMsg`] via the one [`SagaMsg`] trait and
//! stamp a monotonic `t_ms`, so a saga produced either way is identical in shape
//! and obeys the same non-decreasing-`t_ms` invariant.
//!
//! ## No-op-on-release
//! Like everything in the crate, the emit seam is only linked when a host takes
//! `nornir-saga` as a dependency (the idiom is an **optional** dep behind the
//! host's own off-by-default `saga` feature — see `.nornir/saga-idiot-guide.md`).
//! [`Emitter`]/[`UiSink`] are pure `std` transport (no extra deps), and a host
//! that references them only under its `saga` feature ships a clean release.
//!
//! ```
//! use nornir_saga::ui::{Emitter, VecSink, SagaMsg, UiMsg};
//!
//! // The host's Elm Msg → UiMsg projection (implemented once).
//! #[derive(Clone)]
//! enum AppMsg { Nav(String), Tick }
//! impl SagaMsg for AppMsg {
//! fn to_ui_msg(&self) -> Option<UiMsg> {
//! match self { AppMsg::Nav(r) => Some(UiMsg::nav(r)), AppMsg::Tick => None }
//! }
//! }
//!
//! let mut emit = Emitter::new(VecSink::new());
//! emit.tap(&AppMsg::Nav("/home".into())); // projected + stamped + emitted
//! emit.tap(&AppMsg::Tick); // not a movement → dropped
//! let saga = emit.into_sink().into_saga("session");
//! assert_eq!(saga.len(), 1);
//! ```
use std::sync::mpsc::Sender;
use std::time::Instant;
use crate::ui::capture::SagaMsg;
use crate::ui::msg::{Saga, TimedUiMsg, UiMsg};
/// A sink a host feeds its (already-timed) UI movements into. Object-safe, so a
/// caller can hold a `Box<dyn UiSink>` and swap the transport (buffer / channel
/// / null) without touching the emit site.
pub trait UiSink {
/// Take one stamped movement.
fn emit(&mut self, ev: TimedUiMsg);
/// Called once when the emitter is done (flush / close a channel). Default:
/// nothing.
fn close(&mut self) {}
}
/// Drops every movement — the safe default and the release stand-in. Lets a host
/// keep the emit site unconditionally while paying nothing when it wants capture
/// off at runtime.
#[derive(Debug, Default, Clone, Copy)]
pub struct NullSink;
impl UiSink for NullSink {
fn emit(&mut self, _ev: TimedUiMsg) {}
}
/// Collects movements into a `Vec` — the in-process capture + test surface.
/// Drain it into a [`Saga`] with [`into_saga`](VecSink::into_saga).
#[derive(Debug, Default, Clone)]
pub struct VecSink {
/// The movements captured so far, in emit order.
pub events: Vec<TimedUiMsg>,
}
impl VecSink {
/// A fresh, empty sink.
pub fn new() -> VecSink {
VecSink::default()
}
/// Number of movements captured.
pub fn len(&self) -> usize {
self.events.len()
}
/// Whether nothing has been captured.
pub fn is_empty(&self) -> bool {
self.events.is_empty()
}
/// Take the captured movements as a titled [`Saga`] (consumes the sink).
pub fn into_saga(self, title: &str) -> Saga {
Saga { version: Saga::VERSION, title: title.to_string(), events: self.events }
}
}
impl UiSink for VecSink {
fn emit(&mut self, ev: TimedUiMsg) {
self.events.push(ev);
}
}
/// Sends movements down an `mpsc` channel — **the LIVE path**. Pair the
/// receiver with a [`ChannelSource`](crate::ui::ChannelSource) feeding a viewer
/// on another thread, and the movements animate live as the host emits them. A
/// send after the receiver is dropped is silently ignored (the viewer went away;
/// the host must not wedge).
#[derive(Debug, Clone)]
pub struct ChannelSink {
tx: Sender<TimedUiMsg>,
}
impl ChannelSink {
/// Wrap the sending half of a channel whose receiver a viewer pulls.
pub fn new(tx: Sender<TimedUiMsg>) -> ChannelSink {
ChannelSink { tx }
}
}
impl UiSink for ChannelSink {
fn emit(&mut self, ev: TimedUiMsg) {
// Receiver gone → the live viewer closed; drop the movement, never panic.
let _ = self.tx.send(ev);
}
}
/// Stamps projected movements with a monotonic `t_ms` and routes them to a
/// [`UiSink`]. This is the reusable emit engine; the process-global
/// [`tap`](crate::ui::tap) is the zero-config specialization of it (a global
/// `Emitter` over an internal buffer).
///
/// The first emitted movement anchors `t = 0`; every later `t_ms` is
/// milliseconds since that anchor (non-decreasing — the [`Saga`] invariant). For
/// deterministic tests, use [`emit_at`](Emitter::emit_at) to supply `t_ms`
/// explicitly instead of reading the clock.
#[derive(Debug)]
pub struct Emitter<S: UiSink> {
start: Option<Instant>,
sink: S,
}
impl<S: UiSink> Emitter<S> {
/// A new emitter over `sink`, un-anchored (the first emit sets `t = 0`).
pub fn new(sink: S) -> Emitter<S> {
Emitter { start: None, sink }
}
/// Borrow the sink (e.g. to read a [`VecSink`]'s buffer).
pub fn sink(&self) -> &S {
&self.sink
}
/// Mutably borrow the sink.
pub fn sink_mut(&mut self) -> &mut S {
&mut self.sink
}
/// Consume the emitter, returning the sink (e.g. to drain a [`VecSink`]).
pub fn into_sink(self) -> S {
self.sink
}
/// **The tap.** Project a host `Msg` via [`SagaMsg::to_ui_msg`]; if it is a
/// movement, stamp + emit it. Returns `true` when a movement was emitted,
/// `false` when the message was filtered out (a tick / data load).
pub fn tap<M: SagaMsg>(&mut self, msg: &M) -> bool {
match msg.to_ui_msg() {
Some(ui) => {
self.emit(ui);
true
}
None => false,
}
}
/// Stamp a [`UiMsg`] with the monotonic clock and emit it. Prefer
/// [`tap`](Emitter::tap) at the `update` choke point; call this when a
/// component already speaks [`UiMsg`].
pub fn emit(&mut self, msg: UiMsg) {
let now = Instant::now();
let start = *self.start.get_or_insert(now);
let t_ms = now.duration_since(start).as_millis() as u64;
self.sink.emit(TimedUiMsg { t_ms, msg, state_json: None });
}
/// Emit a movement at an explicit `t_ms` (deterministic — no clock read).
/// The emitter is also anchored so any later clock-based [`emit`](Emitter::emit)
/// stays consistent.
pub fn emit_at(&mut self, t_ms: u64, msg: UiMsg) {
self.start.get_or_insert_with(Instant::now);
self.sink.emit(TimedUiMsg { t_ms, msg, state_json: None });
}
/// Emit a movement at an explicit `t_ms` carrying a source `state_json`
/// snapshot the host observed after applying it (the inject-assert law — a
/// replay can check its re-folded model matches).
pub fn emit_at_with_state(&mut self, t_ms: u64, msg: UiMsg, state_json: impl Into<String>) {
self.start.get_or_insert_with(Instant::now);
self.sink.emit(TimedUiMsg { t_ms, msg, state_json: Some(state_json.into()) });
}
/// Finish emitting — flush / close the underlying sink (closes a
/// [`ChannelSink`]'s stream so the LIVE viewer sees end-of-saga).
pub fn close(&mut self) {
self.sink.close();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Clone)]
enum AppMsg {
Nav(&'static str),
Move(f64, f64),
Tick,
}
impl SagaMsg for AppMsg {
fn to_ui_msg(&self) -> Option<UiMsg> {
match self {
AppMsg::Nav(r) => Some(UiMsg::nav(r)),
AppMsg::Move(x, y) => Some(UiMsg::pointer_move(*x, *y)),
AppMsg::Tick => None,
}
}
}
#[test]
fn emitter_taps_projects_stamps_and_filters() {
let mut e = Emitter::new(VecSink::new());
assert!(e.tap(&AppMsg::Nav("/home")));
assert!(!e.tap(&AppMsg::Tick), "non-movement filtered");
assert!(e.tap(&AppMsg::Move(10.0, 20.0)));
let saga = e.into_sink().into_saga("s");
assert_eq!(saga.len(), 2);
assert!(matches!(saga.events[0].msg, UiMsg::Nav { .. }));
assert!(matches!(saga.events[1].msg, UiMsg::PointerMove { .. }));
// t_ms non-decreasing (monotonic clock, first event anchors 0).
assert_eq!(saga.events[0].t_ms, 0);
assert!(saga.events[1].t_ms >= saga.events[0].t_ms);
}
#[test]
fn emit_at_is_deterministic() {
let mut e = Emitter::new(VecSink::new());
e.emit_at(0, UiMsg::nav("/a"));
e.emit_at(120, UiMsg::pointer_move(1.0, 2.0));
e.emit_at_with_state(200, UiMsg::click(1.0, 2.0, "#go"), r#"{"applied":3}"#);
let saga = e.into_sink().into_saga("det");
assert_eq!(saga.duration_ms(), 200);
assert_eq!(saga.events[2].state_json.as_deref(), Some(r#"{"applied":3}"#));
}
#[test]
fn null_sink_drops_everything() {
let mut e = Emitter::new(NullSink);
e.tap(&AppMsg::Nav("/x"));
e.emit(UiMsg::pointer_move(0.0, 0.0));
// NullSink keeps nothing; the emitter still runs.
e.close();
}
#[test]
fn channel_sink_streams_to_a_receiver() {
let (tx, rx) = std::sync::mpsc::channel();
let mut e = Emitter::new(ChannelSink::new(tx));
e.emit_at(0, UiMsg::nav("/live"));
e.emit_at(50, UiMsg::click(5.0, 5.0, "#x"));
e.close();
drop(e); // drop the sender so the receiver stream ends
let got: Vec<_> = rx.into_iter().collect();
assert_eq!(got.len(), 2);
assert!(matches!(got[0].msg, UiMsg::Nav { .. }));
assert!(matches!(got[1].msg, UiMsg::Click { .. }));
}
#[test]
fn channel_sink_ignores_a_dropped_receiver() {
let (tx, rx) = std::sync::mpsc::channel();
drop(rx); // viewer went away
let mut e = Emitter::new(ChannelSink::new(tx));
e.emit_at(0, UiMsg::nav("/x")); // must not panic
}
}