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
15/// Component type.
16#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
17pub enum ComponentType {
18    /// Source.
19    Source,
20
21    /// Relay.
22    Relay,
23
24    /// Decoder.
25    Decoder,
26
27    /// Transform.
28    Transform,
29
30    /// Encoder.
31    Encoder,
32
33    /// Forwarder.
34    Forwarder,
35
36    /// Destination.
37    Destination,
38}
39
40impl ComponentType {
41    /// Returns the string representation of the component type.
42    pub fn as_str(&self) -> &'static str {
43        match self {
44            Self::Source => "source",
45            Self::Relay => "relay",
46            Self::Decoder => "decoder",
47            Self::Transform => "transform",
48            Self::Encoder => "encoder",
49            Self::Forwarder => "forwarder",
50            Self::Destination => "destination",
51        }
52    }
53
54    /// Returns the categorical string representation of the component type.
55    ///
56    /// This is a plural form of the value returned by [`as_str`][Self::as_str]. For example, if [`as_str`][Self::as_str]
57    /// returns `source`, this returns `sources`.
58    pub fn as_category_str(&self) -> &'static str {
59        match self {
60            Self::Source => "sources",
61            Self::Relay => "relays",
62            Self::Decoder => "decoders",
63            Self::Transform => "transforms",
64            Self::Encoder => "encoders",
65            Self::Forwarder => "forwarders",
66            Self::Destination => "destinations",
67        }
68    }
69}
70
71/// A component context.
72///
73/// Holds the identifiers (absolute and relative) for a component, as well as its type.
74#[derive(Clone, Debug, Eq, Hash, PartialEq)]
75pub struct ComponentContext {
76    topology_root: SubsystemIdentifier,
77    component_id: ComponentId,
78    component_type: ComponentType,
79}
80
81impl ComponentContext {
82    /// Creates a new `ComponentContext` rooted at the given topology, with the given identifier and type.
83    pub fn new(topology_root: &SubsystemIdentifier, component_id: ComponentId, component_type: ComponentType) -> Self {
84        Self {
85            topology_root: topology_root.clone(),
86            component_id,
87            component_type,
88        }
89    }
90
91    /// Creates a new `ComponentContext` for a source component with the given identifier, within the named topology.
92    pub fn source(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
93        Self::new(topology_root, component_id, ComponentType::Source)
94    }
95
96    /// Creates a new `ComponentContext` for a relay component with the given identifier, within the named topology.
97    pub fn relay(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
98        Self::new(topology_root, component_id, ComponentType::Relay)
99    }
100
101    /// Creates a new `ComponentContext` for a decoder component with the given identifier, within the named topology.
102    pub fn decoder(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
103        Self::new(topology_root, component_id, ComponentType::Decoder)
104    }
105
106    /// Creates a new `ComponentContext` for a transform component with the given identifier, within the named topology.
107    pub fn transform(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
108        Self::new(topology_root, component_id, ComponentType::Transform)
109    }
110
111    /// Creates a new `ComponentContext` for an encoder component with the given identifier, within the named topology.
112    pub fn encoder(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
113        Self::new(topology_root, component_id, ComponentType::Encoder)
114    }
115
116    /// Creates a new `ComponentContext` for a forwarder component with the given identifier, within the named topology.
117    pub fn forwarder(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
118        Self::new(topology_root, component_id, ComponentType::Forwarder)
119    }
120
121    /// Creates a new `ComponentContext` for a destination component with the given identifier, within the named topology.
122    pub fn destination(topology_root: &SubsystemIdentifier, component_id: ComponentId) -> Self {
123        Self::new(topology_root, component_id, ComponentType::Destination)
124    }
125
126    /// Creates a new `ComponentContext` for a source component with the given identifier, in a test topology.
127    #[cfg(any(test, feature = "test-util"))]
128    pub fn test_source<S: AsRef<str>>(component_id: S) -> Self {
129        Self::source(
130            &SubsystemIdentifier::from_segments(["topology", "test"]),
131            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
132        )
133    }
134
135    /// Creates a new `ComponentContext` for a relay component with the given identifier, in a test topology.
136    #[cfg(any(test, feature = "test-util"))]
137    pub fn test_relay<S: AsRef<str>>(component_id: S) -> Self {
138        Self::relay(
139            &SubsystemIdentifier::from_segments(["topology", "test"]),
140            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
141        )
142    }
143
144    /// Creates a new `ComponentContext` for a decoder component with the given identifier, in a test topology.
145    #[cfg(any(test, feature = "test-util"))]
146    pub fn test_decoder<S: AsRef<str>>(component_id: S) -> Self {
147        Self::decoder(
148            &SubsystemIdentifier::from_segments(["topology", "test"]),
149            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
150        )
151    }
152
153    /// Creates a new `ComponentContext` for a transform component with the given identifier, in a test topology.
154    #[cfg(any(test, feature = "test-util"))]
155    pub fn test_transform<S: AsRef<str>>(component_id: S) -> Self {
156        Self::transform(
157            &SubsystemIdentifier::from_segments(["topology", "test"]),
158            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
159        )
160    }
161
162    /// Creates a new `ComponentContext` for an encoder component with the given identifier, in a test topology.
163    #[cfg(any(test, feature = "test-util"))]
164    pub fn test_encoder<S: AsRef<str>>(component_id: S) -> Self {
165        Self::encoder(
166            &SubsystemIdentifier::from_segments(["topology", "test"]),
167            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
168        )
169    }
170
171    /// Creates a new `ComponentContext` for a forwarder component with the given identifier, in a test topology.
172    #[cfg(any(test, feature = "test-util"))]
173    pub fn test_forwarder<S: AsRef<str>>(component_id: S) -> Self {
174        Self::forwarder(
175            &SubsystemIdentifier::from_segments(["topology", "test"]),
176            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
177        )
178    }
179
180    /// Creates a new `ComponentContext` for a destination component with the given identifier, in a test topology.
181    #[cfg(any(test, feature = "test-util"))]
182    pub fn test_destination<S: AsRef<str>>(component_id: S) -> Self {
183        Self::destination(
184            &SubsystemIdentifier::from_segments(["topology", "test"]),
185            ComponentId::try_from(component_id.as_ref()).expect("invalid component ID"),
186        )
187    }
188
189    /// Returns the component identifier.
190    ///
191    /// This is the relative identifier of the component, unique within its topology but not guaranteed to be globally
192    /// unique. See [`ComponentContext::identity`] for the fully qualified identity.
193    pub fn component_id(&self) -> &ComponentId {
194        &self.component_id
195    }
196
197    /// Returns the component type.
198    pub fn component_type(&self) -> ComponentType {
199        self.component_type
200    }
201
202    /// Returns the fully qualified identity of this component.
203    ///
204    /// The returned identifier uniquely identifies the component within the process, inclusive of the topology to
205    /// which it belongs.
206    pub fn identity(&self) -> SubsystemIdentifier {
207        self.topology_root
208            .clone()
209            .child(self.component_type.as_category_str())
210            .child(&*self.component_id)
211    }
212}
213
214impl fmt::Display for ComponentContext {
215    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
216        write!(f, "{}", self.identity())
217    }
218}
219
220#[cfg(test)]
221mod tests {
222    use std::sync::Arc;
223
224    use saluki_common::sync::shutdown::ShutdownHandle;
225    use tokio::runtime::Handle;
226    use tokio::sync::mpsc;
227
228    use super::decoders::DecoderContext;
229    use super::destinations::DestinationContext;
230    use super::encoders::EncoderContext;
231    use super::forwarders::ForwarderContext;
232    use super::relays::RelayContext;
233    use super::sources::SourceContext;
234    use super::transforms::TransformContext;
235    use super::ComponentContext;
236    use crate::accounting::{ComponentRegistry, MemoryLimiter};
237    use crate::health::{Health, HealthRegistry};
238    use crate::runtime::state::DataspaceRegistry;
239    use crate::runtime::{Supervisor, SupervisorHandle};
240    use crate::support::SubsystemIdentifier;
241    use crate::topology::interconnect::{Consumer, Dispatcher};
242    use crate::topology::{
243        EventsBuffer, EventsConsumer, EventsDispatcher, PayloadsBuffer, PayloadsConsumer, PayloadsDispatcher,
244        TopologyContext,
245    };
246
247    #[test]
248    fn identity_dotted_form() {
249        let context = ComponentContext::test_source("dsd_in");
250        assert_eq!(context.identity().to_string(), "topology.test.sources.dsd_in");
251    }
252
253    #[test]
254    fn context_display_equals_identity_to_string() {
255        let context = ComponentContext::test_transform("dsd_mapper");
256        assert_eq!(context.to_string(), context.identity().to_string());
257    }
258
259    // Double-take handle tests.
260    //
261    // Every `*Context` documents that `take_health_handle` (and, for sources/relays, `take_shutdown_handle`) panics
262    // if the handle has already been taken. Each test below builds a real context, takes the handle once (which must
263    // succeed), then takes it a second time and asserts the documented panic fires with its documented message.
264    //
265    // The shared builders below collapse the otherwise-identical dependency construction across the seven context
266    // types; each context's `new` only differs in which dispatcher/consumer it accepts.
267
268    fn topology_context() -> TopologyContext {
269        TopologyContext::new(
270            Arc::from("test"),
271            MemoryLimiter::noop(),
272            HealthRegistry::new(),
273            Handle::current(),
274            DataspaceRegistry::new(),
275        )
276    }
277
278    fn health_handle() -> Health {
279        HealthRegistry::new()
280            .register_component(&SubsystemIdentifier::from_dotted("test"))
281            .expect("component was not previously registered")
282    }
283
284    fn supervisor_handle() -> SupervisorHandle {
285        Supervisor::new("test").expect("valid supervisor name").handle()
286    }
287
288    fn events_dispatcher(component_context: &ComponentContext) -> EventsDispatcher {
289        Dispatcher::new(component_context.clone())
290    }
291
292    fn payloads_dispatcher(component_context: &ComponentContext) -> PayloadsDispatcher {
293        Dispatcher::new(component_context.clone())
294    }
295
296    fn events_consumer(component_context: &ComponentContext) -> EventsConsumer {
297        let (_tx, rx) = mpsc::channel::<EventsBuffer>(1);
298        Consumer::new(component_context.clone(), rx)
299    }
300
301    fn payloads_consumer(component_context: &ComponentContext) -> PayloadsConsumer {
302        let (_tx, rx) = mpsc::channel::<PayloadsBuffer>(1);
303        Consumer::new(component_context.clone(), rx)
304    }
305
306    fn source_context() -> SourceContext {
307        let cc = ComponentContext::test_source("test");
308        SourceContext::new(
309            &topology_context(),
310            &cc,
311            ComponentRegistry::default(),
312            health_handle(),
313            events_dispatcher(&cc),
314            supervisor_handle(),
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            supervisor_handle(),
327        )
328    }
329
330    fn decoder_context() -> DecoderContext {
331        let cc = ComponentContext::test_decoder("test");
332        DecoderContext::new(
333            &topology_context(),
334            &cc,
335            ComponentRegistry::default(),
336            health_handle(),
337            events_dispatcher(&cc),
338            payloads_consumer(&cc),
339            supervisor_handle(),
340        )
341    }
342
343    fn transform_context() -> TransformContext {
344        let cc = ComponentContext::test_transform("test");
345        TransformContext::new(
346            &topology_context(),
347            &cc,
348            ComponentRegistry::default(),
349            health_handle(),
350            events_dispatcher(&cc),
351            events_consumer(&cc),
352            supervisor_handle(),
353        )
354    }
355
356    fn destination_context() -> DestinationContext {
357        let cc = ComponentContext::test_destination("test");
358        DestinationContext::new(
359            &topology_context(),
360            &cc,
361            ComponentRegistry::default(),
362            health_handle(),
363            events_consumer(&cc),
364            supervisor_handle(),
365        )
366    }
367
368    fn encoder_context() -> EncoderContext {
369        let cc = ComponentContext::test_encoder("test");
370        EncoderContext::new(
371            &topology_context(),
372            &cc,
373            ComponentRegistry::default(),
374            health_handle(),
375            payloads_dispatcher(&cc),
376            events_consumer(&cc),
377            supervisor_handle(),
378        )
379    }
380
381    fn forwarder_context() -> ForwarderContext {
382        let cc = ComponentContext::test_forwarder("test");
383        ForwarderContext::new(
384            &topology_context(),
385            &cc,
386            ComponentRegistry::default(),
387            health_handle(),
388            payloads_consumer(&cc),
389            supervisor_handle(),
390        )
391    }
392
393    // Health-handle double-take: applies to all seven context types.
394
395    #[tokio::test]
396    #[should_panic(expected = "health handle already taken")]
397    async fn source_context_panics_on_double_take_of_health_handle() {
398        let mut ctx = source_context();
399        let _first = ctx.take_health_handle();
400        let _second = ctx.take_health_handle();
401    }
402
403    #[tokio::test]
404    #[should_panic(expected = "health handle already taken")]
405    async fn relay_context_panics_on_double_take_of_health_handle() {
406        let mut ctx = relay_context();
407        let _first = ctx.take_health_handle();
408        let _second = ctx.take_health_handle();
409    }
410
411    #[tokio::test]
412    #[should_panic(expected = "health handle already taken")]
413    async fn decoder_context_panics_on_double_take_of_health_handle() {
414        let mut ctx = decoder_context();
415        let _first = ctx.take_health_handle();
416        let _second = ctx.take_health_handle();
417    }
418
419    #[tokio::test]
420    #[should_panic(expected = "health handle already taken")]
421    async fn transform_context_panics_on_double_take_of_health_handle() {
422        let mut ctx = transform_context();
423        let _first = ctx.take_health_handle();
424        let _second = ctx.take_health_handle();
425    }
426
427    #[tokio::test]
428    #[should_panic(expected = "health handle already taken")]
429    async fn destination_context_panics_on_double_take_of_health_handle() {
430        let mut ctx = destination_context();
431        let _first = ctx.take_health_handle();
432        let _second = ctx.take_health_handle();
433    }
434
435    #[tokio::test]
436    #[should_panic(expected = "health handle already taken")]
437    async fn encoder_context_panics_on_double_take_of_health_handle() {
438        let mut ctx = encoder_context();
439        let _first = ctx.take_health_handle();
440        let _second = ctx.take_health_handle();
441    }
442
443    #[tokio::test]
444    #[should_panic(expected = "health handle already taken")]
445    async fn forwarder_context_panics_on_double_take_of_health_handle() {
446        let mut ctx = forwarder_context();
447        let _first = ctx.take_health_handle();
448        let _second = ctx.take_health_handle();
449    }
450
451    // Shutdown-handle double-take: only sources and relays expose a shutdown handle. The runtime installs it via the
452    // crate-private `set_shutdown_handle` before the component runs, so we do the same here before taking it twice.
453
454    #[tokio::test]
455    #[should_panic(expected = "shutdown handle already taken")]
456    async fn source_context_panics_on_double_take_of_shutdown_handle() {
457        let mut ctx = source_context();
458        ctx.set_shutdown_handle(ShutdownHandle::noop());
459        let _first = ctx.take_shutdown_handle();
460        let _second = ctx.take_shutdown_handle();
461    }
462
463    #[tokio::test]
464    #[should_panic(expected = "shutdown handle already taken")]
465    async fn relay_context_panics_on_double_take_of_shutdown_handle() {
466        let mut ctx = relay_context();
467        ctx.set_shutdown_handle(ShutdownHandle::noop());
468        let _first = ctx.take_shutdown_handle();
469        let _second = ctx.take_shutdown_handle();
470    }
471}
saluki_core/components/mod.rs

saluki_core/components/
mod.rs
