runx_runtime/effects/
registry.rs

1use std::collections::BTreeMap;
2use std::fmt;
3use std::sync::Arc;
4
5use runx_contracts::{Receipt, Reference};
6use runx_parser::GraphStep;
7
8use crate::adapter::SkillOutput;
9
10use super::{
11    EffectAdmission, EffectMetadataRefreshRequest, EffectOutputRequest, EffectReceiptRequest,
12    EffectReplay, EffectReplayOutputRequest, EffectReplayReceiptRequest, EffectStepRequest,
13    RuntimeEffect, RuntimeEffectError,
14};
15
16#[derive(Clone)]
17pub struct RuntimeEffectRegistry {
18    families: BTreeMap<&'static str, Arc<dyn RuntimeEffect>>,
19}
20
21impl RuntimeEffectRegistry {
22    #[must_use]
23    pub fn empty() -> Self {
24        Self {
25            families: BTreeMap::new(),
26        }
27    }
28
29    #[must_use]
30    pub fn with_effect<T>(effect: T) -> Self
31    where
32        T: RuntimeEffect + 'static,
33    {
34        let mut registry = Self::empty();
35        let family = effect.family();
36        registry.families.insert(family, Arc::new(effect));
37        registry
38    }
39
40    pub fn register_effect<T>(&mut self, effect: T) -> Result<(), RuntimeEffectError>
41    where
42        T: RuntimeEffect + 'static,
43    {
44        let family = effect.family();
45        if self.families.contains_key(family) {
46            return Err(RuntimeEffectError::DuplicateFamily {
47                family: family.to_owned(),
48            });
49        }
50        self.families.insert(family, Arc::new(effect));
51        Ok(())
52    }
53
54    pub(crate) fn find_replay(
55        &self,
56        request: EffectStepRequest<'_>,
57    ) -> Result<Option<EffectReplay>, RuntimeEffectError> {
58        for effect in self.families.values() {
59            if let Some(replay) = effect.find_replay(request)? {
60                return Ok(Some(replay));
61            }
62        }
63        Ok(None)
64    }
65
66    pub(crate) fn recover_pending(
67        &self,
68        request: EffectStepRequest<'_>,
69    ) -> Result<(), RuntimeEffectError> {
70        for effect in self.families.values() {
71            effect.recover_pending(request)?;
72        }
73        Ok(())
74    }
75
76    pub(crate) fn admit(
77        &self,
78        request: EffectStepRequest<'_>,
79    ) -> Result<Option<EffectAdmission>, RuntimeEffectError> {
80        for effect in self.families.values() {
81            if let Some(admission) = effect.admit(request)? {
82                return Ok(Some(admission));
83            }
84        }
85        Ok(None)
86    }
87
88    pub(crate) fn prepare_output(
89        &self,
90        request: EffectOutputRequest<'_>,
91    ) -> Result<(), RuntimeEffectError> {
92        let family = request.admission.family();
93        self.require_effect(family)?.prepare_output(request)
94    }
95
96    pub(crate) fn finalize_output(
97        &self,
98        request: EffectReceiptRequest<'_>,
99    ) -> Result<(), RuntimeEffectError> {
100        let family = request.admission.family();
101        self.require_effect(family)?.finalize_output(request)
102    }
103
104    pub(crate) fn persist(
105        &self,
106        request: EffectReceiptRequest<'_>,
107    ) -> Result<(), RuntimeEffectError> {
108        let family = request.admission.family();
109        self.require_effect(family)?.persist(request)
110    }
111
112    pub(crate) fn prepare_replay_output(
113        &self,
114        request: EffectReplayOutputRequest<'_>,
115    ) -> Result<(), RuntimeEffectError> {
116        let family = request.replay.family();
117        self.require_effect(family)?.prepare_replay_output(request)
118    }
119
120    pub(crate) fn validate_replay(
121        &self,
122        request: EffectReplayReceiptRequest<'_>,
123    ) -> Result<(), RuntimeEffectError> {
124        let family = request.replay.family();
125        self.require_effect(family)?.validate_replay(request)
126    }
127
128    pub(crate) fn refresh_output_metadata(
129        &self,
130        output: &mut SkillOutput,
131        receipt: &Receipt,
132    ) -> Result<(), RuntimeEffectError> {
133        for effect in self.families.values() {
134            effect.refresh_output_metadata(EffectMetadataRefreshRequest { output, receipt })?;
135        }
136        Ok(())
137    }
138
139    pub(crate) fn authority_grant_refs(
140        &self,
141        admission: &EffectAdmission,
142    ) -> Result<Vec<Reference>, RuntimeEffectError> {
143        self.require_effect(admission.family())?
144            .authority_grant_refs(admission)
145    }
146
147    pub(crate) fn replay_authority_grant_refs(
148        &self,
149        replay: &EffectReplay,
150    ) -> Result<Vec<Reference>, RuntimeEffectError> {
151        self.require_effect(replay.family())?
152            .replay_authority_grant_refs(replay)
153    }
154
155    pub(crate) fn allows_parallel_step(&self, step: &GraphStep) -> bool {
156        self.families
157            .values()
158            .all(|effect| effect.can_run_parallel(step))
159    }
160
161    fn require_effect(
162        &self,
163        family: &'static str,
164    ) -> Result<&dyn RuntimeEffect, RuntimeEffectError> {
165        self.families.get(family).map(Arc::as_ref).ok_or_else(|| {
166            RuntimeEffectError::MissingFamily {
167                family: family.to_owned(),
168            }
169        })
170    }
171}
172
173impl Default for RuntimeEffectRegistry {
174    fn default() -> Self {
175        Self::empty()
176    }
177}
178
179impl fmt::Debug for RuntimeEffectRegistry {
180    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
181        let families = self.families.keys().copied().collect::<Vec<_>>();
182        formatter
183            .debug_struct("RuntimeEffectRegistry")
184            .field("families", &families)
185            .finish()
186    }
187}
runx_runtime/effects/registry.rs

runx_runtime/effects/
registry.rs
