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

saluki_core/components/
mod.rs
