saluki_core/components/
mod.rs

1//! Component basics.
2
3use std::fmt;
4
5use crate::{support::SubsystemIdentifier, topology::ComponentId};
6
7pub mod decoders;
8pub mod destinations;
9pub mod encoders;
10pub mod forwarders;
11pub mod relays;
12pub mod sources;
13pub mod transforms;
14
15mod build_context;
16pub use self::build_context::BuildContext;
17
18#[cfg(any(test, feature = "test-util"))]
19pub mod test_util;
20
21/// Component type.
22#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
23pub enum ComponentType {
24    /// Source.
25    Source,
26
27    /// Relay.
28    Relay,
29
30    /// Decoder.
31    Decoder,
32
33    /// Transform.
34    Transform,
35
36    /// Encoder.
37    Encoder,
38
39    /// Forwarder.
40    Forwarder,
41
42    /// Destination.
43    Destination,
44}
45
46impl ComponentType {
47    /// Returns the string representation of the component type.
48    pub fn as_str(&self) -> &'static str {
49        match self {
50            Self::Source => "source",
51            Self::Relay => "relay",
52            Self::Decoder => "decoder",
53            Self::Transform => "transform",
54            Self::Encoder => "encoder",
55            Self::Forwarder => "forwarder",
56            Self::Destination => "destination",
57        }
58    }
59
60    /// Returns the categorical string representation of the component type.
61    ///
62    /// This is a plural form of the value returned by [`as_str`][Self::as_str]. For example, if [`as_str`][Self::as_str]
63    /// returns `source`, this returns `sources`.
64    pub fn as_category_str(&self) -> &'static str {
65        match self {
66            Self::Source => "sources",
67            Self::Relay => "relays",
68            Self::Decoder => "decoders",
69            Self::Transform => "transforms",
70            Self::Encoder => "encoders",
71            Self::Forwarder => "forwarders",
72            Self::Destination => "destinations",
73        }
74    }
75}
76
77/// A component context.
78///
79/// Holds the identifiers (absolute and relative) for a component, as well as its type.
80#[derive(Clone, Debug, Eq, Hash, PartialEq)]
81pub struct ComponentContext {
82    topology_root: SubsystemIdentifier,
83    component_id: ComponentId,
84    component_type: ComponentType,
85}
86
87impl ComponentContext {
88    /// Creates a new `ComponentContext` rooted at the given topology, with the given identifier and type.
89    pub fn new(topology_root: &SubsystemIdentifier, component_id: ComponentId, component_type: ComponentType) -> Self {
90        Self {
91            topology_root: topology_root.clone(),
92            component_id,
93            component_type,
94        }
95    }
96
97    /// Creates a new `ComponentContext` for a source component with the given identifier, within the named topology.
98    pub fn source(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
99        Self::new(topology_root, component_id, ComponentType::Source)
100    }
101
102    /// Creates a new `ComponentContext` for a relay component with the given identifier, within the named topology.
103    pub fn relay(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
104        Self::new(topology_root, component_id, ComponentType::Relay)
105    }
106
107    /// Creates a new `ComponentContext` for a decoder component with the given identifier, within the named topology.
108    pub fn decoder(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
109        Self::new(topology_root, component_id, ComponentType::Decoder)
110    }
111
112    /// Creates a new `ComponentContext` for a transform component with the given identifier, within the named topology.
113    pub fn transform(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
114        Self::new(topology_root, component_id, ComponentType::Transform)
115    }
116
117    /// Creates a new `ComponentContext` for an encoder component with the given identifier, within the named topology.
118    pub fn encoder(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
119        Self::new(topology_root, component_id, ComponentType::Encoder)
120    }
121
122    /// Creates a new `ComponentContext` for a forwarder component with the given identifier, within the named topology.
123    pub fn forwarder(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
124        Self::new(topology_root, component_id, ComponentType::Forwarder)
125    }
126
127    /// Creates a new `ComponentContext` for a destination component with the given identifier, within the named topology.
128    pub fn destination(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
129        Self::new(topology_root, component_id, ComponentType::Destination)
130    }
131
132    /// Creates a new `ComponentContext` for a source component with the given identifier, in a test topology.
133    #[cfg(any(test, feature = "test-util"))]
134    pub fn test_source<S: AsRef<str>>(component_id: S) -> Self {
135        Self::source(
136            &SubsystemIdentifier::from_segments(["topology", "test"]),
137            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
138        )
139    }
140
141    /// Creates a new `ComponentContext` for a relay component with the given identifier, in a test topology.
142    #[cfg(any(test, feature = "test-util"))]
143    pub fn test_relay<S: AsRef<str>>(component_id: S) -> Self {
144        Self::relay(
145            &SubsystemIdentifier::from_segments(["topology", "test"]),
146            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
147        )
148    }
149
150    /// Creates a new `ComponentContext` for a decoder component with the given identifier, in a test topology.
151    #[cfg(any(test, feature = "test-util"))]
152    pub fn test_decoder<S: AsRef<str>>(component_id: S) -> Self {
153        Self::decoder(
154            &SubsystemIdentifier::from_segments(["topology", "test"]),
155            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
156        )
157    }
158
159    /// Creates a new `ComponentContext` for a transform component with the given identifier, in a test topology.
160    #[cfg(any(test, feature = "test-util"))]
161    pub fn test_transform<S: AsRef<str>>(component_id: S) -> Self {
162        Self::transform(
163            &SubsystemIdentifier::from_segments(["topology", "test"]),
164            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
165        )
166    }
167
168    /// Creates a new `ComponentContext` for an encoder component with the given identifier, in a test topology.
169    #[cfg(any(test, feature = "test-util"))]
170    pub fn test_encoder<S: AsRef<str>>(component_id: S) -> Self {
171        Self::encoder(
172            &SubsystemIdentifier::from_segments(["topology", "test"]),
173            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
174        )
175    }
176
177    /// Creates a new `ComponentContext` for a forwarder component with the given identifier, in a test topology.
178    #[cfg(any(test, feature = "test-util"))]
179    pub fn test_forwarder<S: AsRef<str>>(component_id: S) -> Self {
180        Self::forwarder(
181            &SubsystemIdentifier::from_segments(["topology", "test"]),
182            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
183        )
184    }
185
186    /// Creates a new `ComponentContext` for a destination component with the given identifier, in a test topology.
187    #[cfg(any(test, feature = "test-util"))]
188    pub fn test_destination<S: AsRef<str>>(component_id: S) -> Self {
189        Self::destination(
190            &SubsystemIdentifier::from_segments(["topology", "test"]),
191            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
192        )
193    }
194
195    /// Returns the component identifier.
196    ///
197    /// This is the relative identifier of the component, unique within its topology but not guaranteed to be globally
198    /// unique. See [`ComponentContext::identity`] for the fully qualified identity.
199    pub fn component_id(&self) -> &ComponentId {
200        &self.component_id
201    }
202
203    /// Returns the component type.
204    pub fn component_type(&self) -> ComponentType {
205        self.component_type
206    }
207
208    /// Returns the fully qualified identity of this component.
209    ///
210    /// The returned identifier uniquely identifies the component within the process, inclusive of the topology to
211    /// which it belongs.
212    pub fn identity(&self) -> SubsystemIdentifier {
213        self.topology_root
214            .clone()
215            .child(self.component_type.as_category_str())
216            .child(&*self.component_id)
217    }
218}
219
220impl fmt::Display for ComponentContext {
221    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
222        write!(f, "{}", self.identity())
223    }
224}
225
226#[cfg(test)]
227mod tests {
228    use super::test_util::TestComponentDriver;
229    use super::ComponentContext;
230
231    #[test]
232    fn identity_dotted_form() {
233        let context = ComponentContext::test_source("dsd_in");
234        assert_eq!(context.identity().to_string(), "topology.test.sources.dsd_in");
235    }
236
237    #[test]
238    fn context_display_equals_identity_to_string() {
239        let context = ComponentContext::test_transform("dsd_mapper");
240        assert_eq!(context.to_string(), context.identity().to_string());
241    }
242
243    // Double-take handle tests.
244    //
245    // Every `*Context` documents that `take_health_handle` (and, for sources/relays, `take_shutdown_handle`) panics
246    // if the handle has already been taken. Each test below builds the context that the component test driver passes
247    // to `run`, takes the handle once (which must succeed), then takes it a second time and asserts the documented
248    // panic fires with its documented message.
249
250    // Health-handle double-take: applies to all seven context types.
251
252    #[tokio::test]
253    #[should_panic(expected = "health handle already taken")]
254    async fn source_context_panics_on_double_take_of_health_handle() {
255        let mut parts = TestComponentDriver::options().source_context(&[]).unwrap();
256        let _first = parts.context.take_health_handle();
257        let _second = parts.context.take_health_handle();
258    }
259
260    #[tokio::test]
261    #[should_panic(expected = "health handle already taken")]
262    async fn relay_context_panics_on_double_take_of_health_handle() {
263        let mut parts = TestComponentDriver::options().relay_context(&[]).unwrap();
264        let _first = parts.context.take_health_handle();
265        let _second = parts.context.take_health_handle();
266    }
267
268    #[tokio::test]
269    #[should_panic(expected = "health handle already taken")]
270    async fn decoder_context_panics_on_double_take_of_health_handle() {
271        let mut parts = TestComponentDriver::options().decoder_context();
272        let _first = parts.context.take_health_handle();
273        let _second = parts.context.take_health_handle();
274    }
275
276    #[tokio::test]
277    #[should_panic(expected = "health handle already taken")]
278    async fn transform_context_panics_on_double_take_of_health_handle() {
279        let mut parts = TestComponentDriver::options().transform_context(&[]).unwrap();
280        let _first = parts.context.take_health_handle();
281        let _second = parts.context.take_health_handle();
282    }
283
284    #[tokio::test]
285    #[should_panic(expected = "health handle already taken")]
286    async fn destination_context_panics_on_double_take_of_health_handle() {
287        let mut parts = TestComponentDriver::options().destination_context();
288        let _first = parts.context.take_health_handle();
289        let _second = parts.context.take_health_handle();
290    }
291
292    #[tokio::test]
293    #[should_panic(expected = "health handle already taken")]
294    async fn encoder_context_panics_on_double_take_of_health_handle() {
295        let mut parts = TestComponentDriver::options().encoder_context();
296        let _first = parts.context.take_health_handle();
297        let _second = parts.context.take_health_handle();
298    }
299
300    #[tokio::test]
301    #[should_panic(expected = "health handle already taken")]
302    async fn forwarder_context_panics_on_double_take_of_health_handle() {
303        let mut parts = TestComponentDriver::options().forwarder_context();
304        let _first = parts.context.take_health_handle();
305        let _second = parts.context.take_health_handle();
306    }
307
308    // Shutdown-handle double-take: only sources and relays expose a shutdown handle, which the context comes with.
309
310    #[tokio::test]
311    #[should_panic(expected = "shutdown handle already taken")]
312    async fn source_context_panics_on_double_take_of_shutdown_handle() {
313        let mut parts = TestComponentDriver::options().source_context(&[]).unwrap();
314        let _first = parts.context.take_shutdown_handle();
315        let _second = parts.context.take_shutdown_handle();
316    }
317
318    #[tokio::test]
319    #[should_panic(expected = "shutdown handle already taken")]
320    async fn relay_context_panics_on_double_take_of_shutdown_handle() {
321        let mut parts = TestComponentDriver::options().relay_context(&[]).unwrap();
322        let _first = parts.context.take_shutdown_handle();
323        let _second = parts.context.take_shutdown_handle();
324    }
325}
saluki_core/components/mod.rs

saluki_core/components/
mod.rs
