saluki_components/transforms/chained/
mod.rs

1use async_trait::async_trait;
2use saluki_common::strings::lower_alphanumeric;
3use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
4use saluki_core::data_model::event::EventType;
5use saluki_core::{
6    components::{transforms::*, ComponentContext},
7    topology::OutputDefinition,
8};
9use saluki_error::GenericError;
10use tokio::select;
11use tracing::{debug, error};
12
13/// Chained transform.
14///
15/// Allows chaining multiple transforms together in a single component, which can avoid the overhead of receiving and
16/// sending events multiple times when concurrency isn't required for processing.
17///
18/// ## Synchronous transforms
19///
20/// This component works with synchronous transforms only. If you need to chain asynchronous transforms, they must be
21/// added to the topology normally.
22#[derive(Default)]
23pub struct ChainedConfiguration {
24    subtransform_builders: Vec<(String, Box<dyn SynchronousTransformBuilder + Send + Sync>)>,
25}
26
27impl ChainedConfiguration {
28    /// Adds a new synchronous transform to the chain.
29    pub fn with_transform_builder<TB>(mut self, subtransform_name: &str, subtransform_builder: TB) -> Self
30    where
31        TB: SynchronousTransformBuilder + Send + Sync + 'static,
32    {
33        let subtransform_id = format!(
34            "{}_{}",
35            self.subtransform_builders.len(),
36            lower_alphanumeric(subtransform_name)
37        );
38        self.subtransform_builders
39            .push((subtransform_id, Box::new(subtransform_builder)));
40        self
41    }
42}
43
44impl MemoryBounds for ChainedConfiguration {
45    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
46        // Capture the size of the heap allocation when the component is built.
47        builder.minimum().with_single_value::<Chained>("component struct");
48
49        for (subtransform_id, subtransform_builder) in self.subtransform_builders.iter() {
50            let mut subtransform_bounds_builder = builder.subcomponent(subtransform_id);
51            subtransform_builder.specify_bounds(&mut subtransform_bounds_builder);
52        }
53    }
54}
55
56#[async_trait]
57impl TransformBuilder for ChainedConfiguration {
58    async fn build(&self, context: ComponentContext) -> Result<Box<dyn Transform + Send>, GenericError> {
59        let mut subtransforms = Vec::new();
60        for (subtransform_id, subtransform_builder) in &self.subtransform_builders {
61            let subtransform = subtransform_builder.build(context.clone()).await?;
62            subtransforms.push((subtransform_id.clone(), subtransform));
63        }
64
65        Ok(Box::new(Chained { subtransforms }))
66    }
67
68    fn input_event_type(&self) -> EventType {
69        EventType::all_bits()
70    }
71
72    fn outputs(&self) -> &[OutputDefinition<EventType>] {
73        static OUTPUTS: &[OutputDefinition<EventType>] = &[OutputDefinition::default_output(EventType::all_bits())];
74        OUTPUTS
75    }
76}
77
78pub struct Chained {
79    subtransforms: Vec<(String, Box<dyn SynchronousTransform + Send>)>,
80}
81
82#[async_trait]
83impl Transform for Chained {
84    async fn run(mut self: Box<Self>, mut context: TransformContext) -> Result<(), GenericError> {
85        let mut health = context.take_health_handle();
86
87        debug!(
88            "Chained transform started with {} synchronous subtransform(s) present.",
89            self.subtransforms.len()
90        );
91
92        // We have to re-associate each subtransform with their allocation group token here, as we don't have access to
93        // it when the bounds are initially defined. Each subtransform's resource group is addressed by its absolute
94        // identity: this component's canonical identity with the subtransform's relative id appended.
95        let component_id = context.component_context().identity();
96        let mut subtransforms = self
97            .subtransforms
98            .into_iter()
99            .map(|(subtransform_id, subtransform)| {
100                (
101                    context
102                        .component_registry()
103                        .get_resource_group_token(&component_id.clone().child(&subtransform_id)),
104                    subtransform,
105                )
106            })
107            .collect::<Vec<_>>();
108
109        health.mark_ready();
110        debug!("Chained transform started.");
111
112        loop {
113            select! {
114                _ = health.live() => continue,
115                maybe_events = context.events().next() => match maybe_events {
116                    Some(mut event_buffer) => {
117                        for (allocation_token, transform) in &mut subtransforms {
118                            let _guard = allocation_token.enter();
119                            transform.transform_buffer(&mut event_buffer);
120                        }
121
122                        if let Err(e) = context.dispatcher().dispatch(event_buffer).await {
123                            error!(error = %e, "Failed to dispatch events.");
124                        }
125                    },
126                    None => break,
127                },
128            }
129        }
130
131        debug!("Chained transform stopped.");
132
133        Ok(())
134    }
135}
saluki_components/transforms/chained/mod.rs

saluki_components/transforms/chained/
mod.rs
