claude
09994b83ed
The three C8 sinks realizing the per-tap subscription model (#283): TapFold (closed FoldKind vocabulary Count|Sum|Mean|Min|Max|First|Last, in-graph accumulation, exactly one finalize row), TapRecorder (bounded SyncSender<(Timestamp, Cell)>, lossless backpressure — a full channel blocks the sim, wall time never state), and TapLive (consumer-owned closure, inline on the sim thread). All three move (Timestamp, Cell) — 16 B, Copy — instead of (Timestamp, Vec<Scalar>): zero per-cycle heap on the tap path (#77 part 2). The shared kind-dispatched newest-cell read lives in tap_cell.rs. Known-good subset commit of iter tap-subscribers (tasks 1-3 of 7, each gated implementer -> spec-compliance -> quality DONE; the loop blocked on a task-4 plan defect, repaired separately): the trace-store streaming writer, the TapPlan wiring seam, and both entry-point rewires follow in this same cycle. Verified directly: cargo test -p aura-std — 109 passed, 0 failed. refs #283, refs #77
89 lines
3.2 KiB
Rust
89 lines
3.2 KiB
Rust
//! `TapLive` — the live-subscription sink (C8) for one declared tap: calls a
|
|
//! consumer-owned closure `(Timestamp, Cell)` inline per fired warm cycle, on
|
|
//! the sim thread, zero alloc. Loss policy is the closure's own (a UI
|
|
//! consumer typically `try_send`s into its own channel and drops when behind)
|
|
//! — per #283: consumer-owned logic and state, no registry ceremony.
|
|
|
|
use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp};
|
|
|
|
use crate::tap_cell::newest_cell;
|
|
|
|
/// A single-column live sink around a consumer closure.
|
|
pub struct TapLive {
|
|
kind: ScalarKind,
|
|
consumer: Box<dyn FnMut(Timestamp, Cell) + Send>,
|
|
}
|
|
|
|
impl TapLive {
|
|
pub fn new(kind: ScalarKind, consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self {
|
|
Self { kind, consumer: Box::new(consumer) }
|
|
}
|
|
}
|
|
|
|
impl Node for TapLive {
|
|
fn lookbacks(&self) -> Vec<usize> {
|
|
vec![1]
|
|
}
|
|
|
|
fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
|
|
let cell = newest_cell(self.kind, &ctx)?; // None until warm
|
|
(self.consumer)(ctx.now(), cell);
|
|
None
|
|
}
|
|
|
|
fn label(&self) -> String {
|
|
"TapLive".to_string()
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use aura_core::{AnyColumn, Scalar};
|
|
use std::sync::mpsc;
|
|
|
|
#[test]
|
|
fn calls_the_consumer_per_warm_cycle_in_order_and_not_before() {
|
|
let (tx, rx) = mpsc::channel();
|
|
let mut sink = TapLive::new(ScalarKind::F64, move |ts, cell| {
|
|
let _ = tx.send((ts, cell.f64()));
|
|
});
|
|
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
|
|
|
|
// cold: consumer not called.
|
|
assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
|
|
assert!(rx.try_recv().is_err(), "no call before warm-up");
|
|
|
|
for (t, v) in [(2_i64, 1.5_f64), (3, 2.5)] {
|
|
inputs[0].push(Scalar::f64(v)).unwrap();
|
|
assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
|
|
}
|
|
let got: Vec<(Timestamp, f64)> = rx.try_iter().collect();
|
|
assert_eq!(got, vec![(Timestamp(2), 1.5), (Timestamp(3), 2.5)]);
|
|
}
|
|
|
|
/// `newest_cell`'s four-kind dispatch (shared with `TapFold`) is
|
|
/// otherwise only exercised at F64/I64; the Bool and Timestamp arms
|
|
/// need a witness too, so a regression in either goes caught.
|
|
#[test]
|
|
fn calls_the_consumer_at_bool_and_timestamp_kinds_too() {
|
|
let (tx, rx) = mpsc::channel();
|
|
let mut sink = TapLive::new(ScalarKind::Bool, move |ts, cell| {
|
|
let _ = tx.send((ts, cell.bool()));
|
|
});
|
|
let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::Bool, 1)];
|
|
inputs[0].push(Scalar::bool(true)).unwrap();
|
|
assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
|
|
assert_eq!(rx.try_recv(), Ok((Timestamp(1), true)));
|
|
|
|
let (tx2, rx2) = mpsc::channel();
|
|
let mut sink2 = TapLive::new(ScalarKind::Timestamp, move |ts, cell| {
|
|
let _ = tx2.send((ts, cell.ts()));
|
|
});
|
|
let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::Timestamp, 1)];
|
|
inputs2[0].push(Scalar::ts(Timestamp(99))).unwrap();
|
|
assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(1))), None);
|
|
assert_eq!(rx2.try_recv(), Ok((Timestamp(1), Timestamp(99))));
|
|
}
|
|
}
|