khora_data/flow/
registration.rs

1// Copyright 2025 eraflo
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Auto-registration of [`Flow`](super::Flow) implementations via `inventory`.
16
17use khora_core::lane::LaneBus;
18use khora_core::Runtime;
19
20use crate::ecs::{SemanticDomain, World};
21
22/// Registration entry for a [`Flow`](super::Flow) — submitted by each
23/// concrete Flow implementation via [`inventory::submit!`].
24///
25/// Type erasure: because [`Flow`](super::Flow) has an associated `View`
26/// type, we cannot store trait objects directly. Each Flow provides a
27/// trampoline `run` function that owns its lifecycle (typically delegating
28/// to a `static OnceLock<Mutex<MyFlow>>` instance) and publishes its View
29/// into the [`LaneBus`].
30pub struct FlowRegistration {
31    /// Stable identifier — matches `Flow::NAME`.
32    pub name: &'static str,
33    /// Domain this Flow serves — matches `Flow::DOMAIN`.
34    pub domain: SemanticDomain,
35    /// Trampoline that runs select + project and publishes the View.
36    pub run: fn(&mut World, &mut LaneBus, &Runtime),
37}
38
39inventory::collect!(FlowRegistration);
40
41/// One Substrate-Pass execution of `flow`, with view-cache support. This is
42/// the body of the [`register_flow!`] trampoline, extracted so the caching
43/// behaviour is a single, directly testable code path.
44///
45/// `cache` holds the `(key, view)` pair of the previously published view.
46/// When [`Flow::cache_key`] returns `Some(k)` equal to the cached key, the
47/// cached view is republished (cloned) without re-running
48/// `select`/`project`; otherwise the flow projects normally and the result
49/// replaces the cache (only when caching is enabled — a `None` key clears
50/// it, so a flow can opt out dynamically without risking a stale entry).
51pub fn run_flow_cached<F: super::Flow>(
52    flow: &mut F,
53    cache: &mut Option<(u64, F::View)>,
54    world: &mut World,
55    bus: &mut LaneBus,
56    runtime: &Runtime,
57) {
58    let key = flow.cache_key(world, runtime);
59    if let (Some(k), Some((cached_key, cached_view))) = (key, cache.as_ref()) {
60        if k == *cached_key {
61            bus.publish(cached_view.clone());
62            return;
63        }
64    }
65
66    let sel = flow.select(world, runtime);
67    let view = flow.project(world, &sel, runtime);
68    *cache = key.map(|k| (k, view.clone()));
69    bus.publish(view);
70}
71
72/// Convenience macro for declaring a Flow registration with the standard
73/// trampoline (single static instance, no per-frame allocation).
74///
75/// # Example
76///
77/// ```rust,ignore
78/// use khora_data::flow::register_flow;
79///
80/// pub struct MyFlow { /* ... */ }
81/// impl khora_data::flow::Flow for MyFlow { /* ... */ }
82///
83/// register_flow!(MyFlow);
84/// ```
85#[macro_export]
86macro_rules! register_flow {
87    ($flow_ty:ty) => {
88        const _: () = {
89            fn run_flow(
90                world: &mut $crate::ecs::World,
91                bus: &mut khora_core::lane::LaneBus,
92                runtime: &khora_core::Runtime,
93            ) {
94                use std::sync::{Mutex, OnceLock};
95                // The flow instance and its view cache live together behind
96                // the same lock: the cached `(key, view)` of the previously
97                // published view, reused when `Flow::cache_key` matches.
98                type FlowState = (
99                    $flow_ty,
100                    Option<(u64, <$flow_ty as $crate::flow::Flow>::View)>,
101                );
102                static INSTANCE: OnceLock<Mutex<FlowState>> = OnceLock::new();
103                let mut state = INSTANCE
104                    .get_or_init(|| Mutex::new((<$flow_ty as Default>::default(), None)))
105                    .lock()
106                    .expect("Flow mutex poisoned");
107                let (flow, cache) = &mut *state;
108                $crate::flow::run_flow_cached(flow, cache, world, bus, runtime);
109            }
110            inventory::submit! {
111                $crate::flow::FlowRegistration {
112                    name: <$flow_ty as $crate::flow::Flow>::NAME,
113                    domain: <$flow_ty as $crate::flow::Flow>::DOMAIN,
114                    run: run_flow,
115                }
116            }
117        };
118    };
119}
120
121#[cfg(test)]
122mod tests {
123    use std::sync::atomic::{AtomicUsize, Ordering};
124
125    use khora_core::lane::LaneBus;
126    use khora_core::Runtime;
127
128    use crate::ecs::{AudioListener, AudioSource, SemanticDomain, World};
129    use crate::flow::{combine_cache_key, AudioFlow, AudioView, Flow, Selection};
130
131    /// Minimal view carrying the projection count of the flow that built it.
132    #[derive(Debug, Clone, PartialEq)]
133    struct CountedView {
134        projections: usize,
135    }
136
137    /// Cached test flow keyed on the Audio + Spatial epochs, mirroring the
138    /// real `AudioFlow` key; counts how many times `project` actually runs.
139    #[derive(Default)]
140    struct CountedFlow {
141        projections: AtomicUsize,
142    }
143
144    impl Flow for CountedFlow {
145        type View = CountedView;
146
147        const DOMAIN: SemanticDomain = SemanticDomain::Audio;
148        const NAME: &'static str = "counted_test";
149
150        fn cache_key(&self, world: &World, _runtime: &Runtime) -> Option<u64> {
151            Some(combine_cache_key([
152                world.instance_id(),
153                world.domain_epoch(SemanticDomain::Audio),
154                world.domain_epoch(SemanticDomain::Spatial),
155            ]))
156        }
157
158        fn project(&self, _world: &World, _sel: &Selection, _runtime: &Runtime) -> CountedView {
159            CountedView {
160                projections: self.projections.fetch_add(1, Ordering::Relaxed) + 1,
161            }
162        }
163    }
164
165    /// Flow that keeps the default `cache_key` (`None`) — never cached.
166    #[derive(Default)]
167    struct UncachedFlow {
168        projections: AtomicUsize,
169    }
170
171    impl Flow for UncachedFlow {
172        type View = CountedView;
173
174        const DOMAIN: SemanticDomain = SemanticDomain::Audio;
175        const NAME: &'static str = "uncached_test";
176
177        fn project(&self, _world: &World, _sel: &Selection, _runtime: &Runtime) -> CountedView {
178            CountedView {
179                projections: self.projections.fetch_add(1, Ordering::Relaxed) + 1,
180            }
181        }
182    }
183
184    #[test]
185    fn cached_flow_republishes_without_reprojecting_when_world_is_unchanged() {
186        let mut world = World::new();
187        let runtime = Runtime::new();
188        let mut flow = CountedFlow::default();
189        let mut cache = None;
190
191        let mut bus = LaneBus::new();
192        super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
193        assert_eq!(flow.projections.load(Ordering::Relaxed), 1);
194        assert_eq!(
195            bus.get::<CountedView>(),
196            Some(&CountedView { projections: 1 })
197        );
198
199        // Nothing mutated the world: the second run must republish the
200        // cached view instead of projecting again.
201        let mut bus = LaneBus::new();
202        super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
203        assert_eq!(
204            flow.projections.load(Ordering::Relaxed),
205            1,
206            "unchanged world must not re-run project"
207        );
208        assert_eq!(
209            bus.get::<CountedView>(),
210            Some(&CountedView { projections: 1 }),
211            "the cached view must still be published into the fresh bus"
212        );
213    }
214
215    #[test]
216    fn cached_flow_reprojects_after_a_domain_mutation() {
217        let mut world = World::new();
218        let runtime = Runtime::new();
219        let mut flow = CountedFlow::default();
220        let mut cache = None;
221
222        let mut bus = LaneBus::new();
223        super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
224        assert_eq!(flow.projections.load(Ordering::Relaxed), 1);
225
226        // Mutate the Audio domain — the key changes, the cache must miss.
227        world.spawn(AudioListener);
228
229        let mut bus = LaneBus::new();
230        super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
231        assert_eq!(
232            flow.projections.load(Ordering::Relaxed),
233            2,
234            "a mutated domain must invalidate the cached view"
235        );
236        assert_eq!(
237            bus.get::<CountedView>(),
238            Some(&CountedView { projections: 2 })
239        );
240    }
241
242    #[test]
243    fn uncached_flow_reprojects_every_run() {
244        let mut world = World::new();
245        let runtime = Runtime::new();
246        let mut flow = UncachedFlow::default();
247        let mut cache = None;
248
249        for expected in 1..=3 {
250            let mut bus = LaneBus::new();
251            super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
252            assert_eq!(flow.projections.load(Ordering::Relaxed), expected);
253            assert!(
254                cache.is_none(),
255                "a `None` key must never populate the cache"
256            );
257        }
258    }
259
260    #[test]
261    fn audio_flow_is_never_stale_after_an_audio_source_insert() {
262        let mut world = World::new();
263        let runtime = Runtime::new();
264        let mut flow = AudioFlow;
265        let mut cache = None;
266
267        // Prime the cache with an empty scene.
268        let mut bus = LaneBus::new();
269        super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
270        let view = bus.get::<AudioView>().expect("AudioFlow publishes a view");
271        assert_eq!(view.source_count, 0);
272
273        // Insert an AudioSource: the next run must re-project, not serve
274        // the cached (now outdated) view.
275        world.spawn(AudioSource::default());
276
277        let mut bus = LaneBus::new();
278        super::run_flow_cached(&mut flow, &mut cache, &mut world, &mut bus, &runtime);
279        let view = bus.get::<AudioView>().expect("AudioFlow publishes a view");
280        assert_eq!(
281            view.source_count, 1,
282            "the published view must reflect the inserted AudioSource"
283        );
284    }
285}
khora_data/flow/registration.rs

khora_data/flow/
registration.rs
