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

vector/transforms/
mod.rs
