proxima_primitives/pipe/
source.rs

1//! [`SourcePipe`] is the background-producer face of the [`SendPipe`] root
2//! form: a `Signal -> ()` pipe with `Err` pinned to [`ProximaError`]. Where
3//! [`Handler`](crate::pipe::handler::Handler) serves one request and returns one
4//! response, a `SourcePipe` runs an unbounded loop that observes a
5//! cancellation [`Signal`] cooperatively and returns once it has drained.
6//!
7//! `IntervalPipe` and `ScheduledTriggerPipe` are the canonical implementors:
8//! each wraps a `run_*_loop(period, inner, factory, cancel)` free function
9//! that already takes a `Signal` and only returns when the signal fires.
10//! `SourcePipe` is a blanket impl over the right `SendPipe` shape — nothing
11//! to implement directly.
12//!
13//! [`SourceHandle`] is the erased, shareable form [`App::source`](crate)
14//! stores; [`ProducerLifecycle`](crate::pipe::lifecycle::ProducerLifecycle) drives
15//! every registered source cooperatively via `spawn_from_source`.
16
17use serde_json::Value;
18
19use proxima_core::ProximaError;
20use proxima_core::factory::Named;
21use proxima_core::signal::Signal;
22
23use crate::pipe::alloc_tier;
24use crate::pipe::primitives::SendPipe;
25
26/// The background-producer face of [`SendPipe`]: `Signal -> ()`, `Err`
27/// pinned to [`ProximaError`]. Blanket-implemented for every qualifying
28/// `SendPipe` — nothing to implement directly.
29pub trait SourcePipe: SendPipe<In = Signal, Out = (), Err = ProximaError> {}
30
31impl<Implementor> SourcePipe for Implementor where
32    Implementor: SendPipe<In = Signal, Out = (), Err = ProximaError>
33{
34}
35
36/// Erased, shareable source handle — the runtime-dispatch alias
37/// `Arc<dyn SendDynPipe<Signal, ()>>` that `App::source` stores.
38pub type SourceHandle = alloc_tier::PipeHandle<Signal, ()>;
39
40/// Erase a [`SourcePipe`] into a shareable [`SourceHandle`].
41pub fn into_source_handle<Implementor>(pipe: Implementor) -> SourceHandle
42where
43    Implementor: SendPipe<In = Signal, Out = (), Err = ProximaError> + 'static,
44{
45    alloc_tier::into_handle(pipe)
46}
47
48/// Config-driven source factory: the [`SourcePipe`] analogue of
49/// [`PipeFactory`](crate::pipe::pipe_factory::PipeFactory). `IntervalPipe` and
50/// `ScheduledTriggerPipe` each register a thin `SourceFactory` so
51/// `App::apply_settings`'s producer-graph-config step resolves a TOML row
52/// straight into a [`SourceHandle`] via the shared
53/// [`FactoryRegistry`](proxima_core::factory::FactoryRegistry) primitive.
54pub trait SourceFactory: Named {
55    /// Build a source from a config spec.
56    ///
57    /// # Errors
58    ///
59    /// Returns `Err` when the spec fails to parse or lower into a source.
60    fn build(&self, spec: &Value) -> Result<SourceHandle, ProximaError>;
61}
62
63/// The source registry — [`proxima_core::FactoryRegistry`] specialized to
64/// `dyn SourceFactory`, mirroring
65/// [`PipeFactoryRegistry`](crate::pipe::pipe_factory::PipeFactoryRegistry).
66pub type SourceFactoryRegistry = proxima_core::FactoryRegistry<dyn SourceFactory>;
67
68// `#[proxima::test]` pulls in the `proxima` dev-dependency, which the
69// loom build keeps out of the graph (see
70// `[target.'cfg(not(loom))'.dev-dependencies]` in Cargo.toml); these
71// tests are unrelated to the Notify/watch loom protocol.
72#[cfg(all(test, not(loom)))]
73#[allow(clippy::unwrap_used, clippy::expect_used)]
74mod tests {
75    use super::*;
76    use core::future::Future;
77    use core::sync::atomic::{AtomicUsize, Ordering};
78
79    struct CountingSource {
80        calls: alloc::sync::Arc<AtomicUsize>,
81    }
82
83    impl SendPipe for CountingSource {
84        type In = Signal;
85        type Out = ();
86        type Err = ProximaError;
87
88        fn call(&self, _cancel: Signal) -> impl Future<Output = Result<(), ProximaError>> + Send {
89            self.calls.fetch_add(1, Ordering::SeqCst);
90            async move { Ok(()) }
91        }
92    }
93
94    #[proxima::test]
95    async fn source_handle_dispatches_to_inner() {
96        let calls = alloc::sync::Arc::new(AtomicUsize::new(0));
97        let handle: SourceHandle = into_source_handle(CountingSource {
98            calls: calls.clone(),
99        });
100        let signal = Signal::new();
101        SendPipe::call(&handle, signal).await.expect("call");
102        assert_eq!(calls.load(Ordering::SeqCst), 1);
103    }
104}
proxima_primitives/pipe/source.rs

proxima_primitives/pipe/
source.rs
