vector/transforms/
mod.rs

1#![allow(missing_docs)]
2#[allow(unused_imports)]
3use std::collections::HashSet;
4
5pub mod dedupe;
6pub mod reduce;
7#[cfg(feature = "transforms-impl-sample")]
8pub mod sample;
9
10#[cfg(feature = "transforms-aggregate")]
11pub mod aggregate;
12#[cfg(feature = "transforms-aws_ec2_metadata")]
13pub mod aws_ec2_metadata;
14#[cfg(feature = "transforms-exclusive-route")]
15mod exclusive_route;
16#[cfg(feature = "transforms-filter")]
17pub mod filter;
18#[cfg(feature = "transforms-incremental_to_absolute")]
19pub mod incremental_to_absolute;
20#[cfg(feature = "transforms-log_to_metric")]
21pub mod log_to_metric;
22#[cfg(feature = "transforms-lua")]
23pub mod lua;
24#[cfg(feature = "transforms-metric_to_log")]
25pub mod metric_to_log;
26#[cfg(feature = "transforms-remap")]
27pub mod remap;
28#[cfg(feature = "transforms-route")]
29pub mod route;
30#[cfg(feature = "transforms-tag_cardinality_limit")]
31pub mod tag_cardinality_limit;
32#[cfg(feature = "transforms-throttle")]
33pub mod throttle;
34#[cfg(feature = "transforms-window")]
35pub mod window;
36
37pub use vector_lib::transform::{
38    FunctionTransform, OutputBuffer, SyncTransform, TaskTransform, Transform, TransformOutputs,
39    TransformOutputsBuf,
40};
41
42#[cfg(test)]
43mod test {
44    use futures::Stream;
45    use futures_util::SinkExt;
46    use tokio::sync::mpsc;
47    use tokio_util::sync::PollSender;
48    use vector_lib::transform::FunctionTransform;
49
50    use crate::{
51        config::{
52            ConfigBuilder, TransformConfig,
53            unit_test::{UnitTestStreamSinkConfig, UnitTestStreamSourceConfig},
54        },
55        event::Event,
56        test_util::start_topology,
57        topology::RunningTopology,
58        transforms::OutputBuffer,
59    };
60
61    /// Transform a single `Event` through the `FunctionTransform`
62    ///
63    /// # Panics
64    ///
65    /// If `ft` attempts to emit more than one `Event` on transform this
66    /// function will panic.
67    // We allow dead_code here to avoid unused warnings when we compile our
68    // benchmarks as tests. It's a valid warning -- the benchmarks don't use
69    // this function -- but flagging this function off for bench flags will
70    // issue a unused warnings about the import above.
71    #[allow(dead_code)]
72    pub fn transform_one(ft: &mut dyn FunctionTransform, event: Event) -> Option<Event> {
73        let mut buf = OutputBuffer::with_capacity(1);
74        ft.transform(&mut buf, event);
75        assert!(buf.len() <= 1);
76        buf.into_events().next()
77    }
78
79    #[allow(dead_code)]
80    pub async fn create_topology<T: TransformConfig + 'static>(
81        events: impl Stream<Item = Event> + Send + 'static,
82        transform_config: T,
83    ) -> (RunningTopology, mpsc::Receiver<Event>) {
84        let mut builder = ConfigBuilder::default();
85
86        let (tx, rx) = mpsc::channel(1);
87
88        // TODO: Use non-hard-coded names to improve tests.
89        builder.add_source("in", UnitTestStreamSourceConfig::new(events));
90        builder.add_transform("transform", &["in"], transform_config);
91        builder.add_sink(
92            "out",
93            &["transform"],
94            UnitTestStreamSinkConfig::new(
95                PollSender::new(tx).sink_map_err(|error| panic!("{}", error)),
96            ),
97        );
98
99        let config = builder.build().expect("building config should not fail");
100        let (topology, _) = start_topology(config, false).await;
101
102        (topology, rx)
103    }
104}
vector/transforms/mod.rs

vector/transforms/
mod.rs
