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 std::sync::Arc;
229
230    use saluki_common::sync::shutdown::ShutdownHandle;
231    use tokio::runtime::Handle;
232    use tokio::sync::mpsc;
233
234    use super::decoders::DecoderContext;
235    use super::destinations::DestinationContext;
236    use super::encoders::EncoderContext;
237    use super::forwarders::ForwarderContext;
238    use super::relays::RelayContext;
239    use super::sources::SourceContext;
240    use super::transforms::TransformContext;
241    use super::ComponentContext;
242    use crate::accounting::{ComponentRegistry, MemoryLimiter};
243    use crate::health::{Health, HealthRegistry};
244    use crate::runtime::state::DataspaceRegistry;
245    use crate::support::SubsystemIdentifier;
246    use crate::topology::interconnect::{Consumer, Dispatcher};
247    use crate::topology::{
248        EventsBuffer, EventsConsumer, EventsDispatcher, PayloadsBuffer, PayloadsConsumer, PayloadsDispatcher,
249        TopologyContext,
250    };
251
252    #[test]
253    fn identity_dotted_form() {
254        let context = ComponentContext::test_source("dsd_in");
255        assert_eq!(context.identity().to_string(), "topology.test.sources.dsd_in");
256    }
257
258    #[test]
259    fn context_display_equals_identity_to_string() {
260        let context = ComponentContext::test_transform("dsd_mapper");
261        assert_eq!(context.to_string(), context.identity().to_string());
262    }
263
264    // Double-take handle tests.
265    //
266    // Every `*Context` documents that `take_health_handle` (and, for sources/relays, `take_shutdown_handle`) panics
267    // if the handle has already been taken. Each test below builds a real context, takes the handle once (which must
268    // succeed), then takes it a second time and asserts the documented panic fires with its documented message.
269    //
270    // The shared builders below collapse the otherwise-identical dependency construction across the seven context
271    // types; each context's `new` only differs in which dispatcher/consumer it accepts.
272
273    fn topology_context() -> TopologyContext {
274        TopologyContext::new(
275            Arc::from("test"),
276            MemoryLimiter::noop(),
277            HealthRegistry::new(),
278            Handle::current(),
279            DataspaceRegistry::new(),
280        )
281    }
282
283    fn health_handle() -> Health {
284        HealthRegistry::new()
285            .register_component(&SubsystemIdentifier::from_dotted("test"))
286            .expect("component was not previously registered")
287    }
288
289    fn events_dispatcher(component_context: &ComponentContext) -> EventsDispatcher {
290        Dispatcher::new(component_context.clone())
291    }
292
293    fn payloads_dispatcher(component_context: &ComponentContext) -> PayloadsDispatcher {
294        Dispatcher::new(component_context.clone())
295    }
296
297    fn events_consumer(component_context: &ComponentContext) -> EventsConsumer {
298        let (_tx, rx) = mpsc::channel::<EventsBuffer>(1);
299        Consumer::new(component_context.clone(), rx)
300    }
301
302    fn payloads_consumer(component_context: &ComponentContext) -> PayloadsConsumer {
303        let (_tx, rx) = mpsc::channel::<PayloadsBuffer>(1);
304        Consumer::new(component_context.clone(), rx)
305    }
306
307    fn source_context() -> SourceContext {
308        let cc = ComponentContext::test_source("test");
309        SourceContext::new(
310            &topology_context(),
311            &cc,
312            ComponentRegistry::default(),
313            health_handle(),
314            events_dispatcher(&cc),
315        )
316    }
317
318    fn relay_context() -> RelayContext {
319        let cc = ComponentContext::test_relay("test");
320        RelayContext::new(
321            &topology_context(),
322            &cc,
323            ComponentRegistry::default(),
324            health_handle(),
325            payloads_dispatcher(&cc),
326        )
327    }
328
329    fn decoder_context() -> DecoderContext {
330        let cc = ComponentContext::test_decoder("test");
331        DecoderContext::new(
332            &topology_context(),
333            &cc,
334            ComponentRegistry::default(),
335            health_handle(),
336            events_dispatcher(&cc),
337            payloads_consumer(&cc),
338        )
339    }
340
341    fn transform_context() -> TransformContext {
342        let cc = ComponentContext::test_transform("test");
343        TransformContext::new(
344            &topology_context(),
345            &cc,
346            ComponentRegistry::default(),
347            health_handle(),
348            events_dispatcher(&cc),
349            events_consumer(&cc),
350        )
351    }
352
353    fn destination_context() -> DestinationContext {
354        let cc = ComponentContext::test_destination("test");
355        DestinationContext::new(
356            &topology_context(),
357            &cc,
358            ComponentRegistry::default(),
359            health_handle(),
360            events_consumer(&cc),
361        )
362    }
363
364    fn encoder_context() -> EncoderContext {
365        let cc = ComponentContext::test_encoder("test");
366        EncoderContext::new(
367            &topology_context(),
368            &cc,
369            ComponentRegistry::default(),
370            health_handle(),
371            payloads_dispatcher(&cc),
372            events_consumer(&cc),
373        )
374    }
375
376    fn forwarder_context() -> ForwarderContext {
377        let cc = ComponentContext::test_forwarder("test");
378        ForwarderContext::new(
379            &topology_context(),
380            &cc,
381            ComponentRegistry::default(),
382            health_handle(),
383            payloads_consumer(&cc),
384        )
385    }
386
387    // Health-handle double-take: applies to all seven context types.
388
389    #[tokio::test]
390    #[should_panic(expected = "health handle already taken")]
391    async fn source_context_panics_on_double_take_of_health_handle() {
392        let mut ctx = source_context();
393        let _first = ctx.take_health_handle();
394        let _second = ctx.take_health_handle();
395    }
396
397    #[tokio::test]
398    #[should_panic(expected = "health handle already taken")]
399    async fn relay_context_panics_on_double_take_of_health_handle() {
400        let mut ctx = relay_context();
401        let _first = ctx.take_health_handle();
402        let _second = ctx.take_health_handle();
403    }
404
405    #[tokio::test]
406    #[should_panic(expected = "health handle already taken")]
407    async fn decoder_context_panics_on_double_take_of_health_handle() {
408        let mut ctx = decoder_context();
409        let _first = ctx.take_health_handle();
410        let _second = ctx.take_health_handle();
411    }
412
413    #[tokio::test]
414    #[should_panic(expected = "health handle already taken")]
415    async fn transform_context_panics_on_double_take_of_health_handle() {
416        let mut ctx = transform_context();
417        let _first = ctx.take_health_handle();
418        let _second = ctx.take_health_handle();
419    }
420
421    #[tokio::test]
422    #[should_panic(expected = "health handle already taken")]
423    async fn destination_context_panics_on_double_take_of_health_handle() {
424        let mut ctx = destination_context();
425        let _first = ctx.take_health_handle();
426        let _second = ctx.take_health_handle();
427    }
428
429    #[tokio::test]
430    #[should_panic(expected = "health handle already taken")]
431    async fn encoder_context_panics_on_double_take_of_health_handle() {
432        let mut ctx = encoder_context();
433        let _first = ctx.take_health_handle();
434        let _second = ctx.take_health_handle();
435    }
436
437    #[tokio::test]
438    #[should_panic(expected = "health handle already taken")]
439    async fn forwarder_context_panics_on_double_take_of_health_handle() {
440        let mut ctx = forwarder_context();
441        let _first = ctx.take_health_handle();
442        let _second = ctx.take_health_handle();
443    }
444
445    // Shutdown-handle double-take: only sources and relays expose a shutdown handle. The runtime installs it via the
446    // crate-private `set_shutdown_handle` before the component runs, so we do the same here before taking it twice.
447
448    #[tokio::test]
449    #[should_panic(expected = "shutdown handle already taken")]
450    async fn source_context_panics_on_double_take_of_shutdown_handle() {
451        let mut ctx = source_context();
452        ctx.set_shutdown_handle(ShutdownHandle::noop());
453        let _first = ctx.take_shutdown_handle();
454        let _second = ctx.take_shutdown_handle();
455    }
456
457    #[tokio::test]
458    #[should_panic(expected = "shutdown handle already taken")]
459    async fn relay_context_panics_on_double_take_of_shutdown_handle() {
460        let mut ctx = relay_context();
461        ctx.set_shutdown_handle(ShutdownHandle::noop());
462        let _first = ctx.take_shutdown_handle();
463        let _second = ctx.take_shutdown_handle();
464    }
465}
saluki_core/components/mod.rs

saluki_core/components/
mod.rs
