vector/sinks/util/
mod.rs

1pub mod adaptive_concurrency;
2pub mod auth;
3// https://github.com/mcarton/rust-derivative/issues/112
4#[allow(clippy::non_canonical_clone_impl)]
5pub mod batch;
6pub mod buffer;
7pub mod builder;
8pub mod compressor;
9pub mod datagram;
10pub mod encoding;
11pub mod http;
12pub mod metadata;
13pub mod normalizer;
14pub mod partitioner;
15pub mod path_confinement;
16pub mod processed_event;
17pub mod request_builder;
18pub mod retries;
19pub mod service;
20pub mod sink;
21pub mod snappy;
22pub mod socket_bytes_sink;
23pub mod statistic;
24pub mod tcp;
25#[cfg(any(test, feature = "test-utils"))]
26pub mod test;
27pub mod udp;
28#[cfg(unix)]
29pub mod unix;
30pub mod uri;
31pub mod zstd;
32
33use std::borrow::Cow;
34
35pub use batch::{
36    Batch, BatchConfig, BatchSettings, BatchSize, BulkSizeBasedDefaultBatchSettings, Merged,
37    NoDefaultsBatchSettings, PushResult, RealtimeEventBasedDefaultBatchSettings,
38    RealtimeSizeBasedDefaultBatchSettings, SinkBatchSettings, Unmerged,
39};
40pub use buffer::{
41    Buffer, Compression, PartitionBuffer, PartitionInnerBuffer,
42    json::{BoxedRawValue, JsonArrayBuffer},
43    partition::Partition,
44    vec::{EncodedLength, VecBuffer},
45};
46pub use builder::SinkBuilderExt;
47use chrono::{FixedOffset, Offset, Utc};
48pub use compressor::Compressor;
49pub use normalizer::Normalizer;
50pub use request_builder::{IncrementalRequestBuilder, RequestBuilder};
51pub use service::{
52    Concurrency, ServiceBuilderExt, TowerBatchedSink, TowerPartitionSink, TowerRequestConfig,
53    TowerRequestLayer, TowerRequestSettings,
54};
55pub use sink::{BatchSink, PartitionBatchSink, StreamSink};
56use snafu::Snafu;
57pub use uri::UriSerde;
58use vector_lib::{TimeZone, json_size::JsonSize};
59
60use crate::event::EventFinalizers;
61
62#[derive(Debug, Snafu)]
63enum SinkBuildError {
64    #[snafu(display("Missing host in address field"))]
65    MissingHost,
66    #[snafu(display("Missing port in address field"))]
67    MissingPort,
68}
69
70#[derive(Debug)]
71pub struct EncodedEvent<I> {
72    pub item: I,
73    pub finalizers: EventFinalizers,
74    pub byte_size: usize,
75    pub json_byte_size: JsonSize,
76}
77
78impl<I> EncodedEvent<I> {
79    /// Create a trivial input with no metadata. This method will be
80    /// removed when all sinks are converted.
81    pub fn new(item: I, byte_size: usize, json_byte_size: JsonSize) -> Self {
82        Self {
83            item,
84            finalizers: Default::default(),
85            byte_size,
86            json_byte_size,
87        }
88    }
89
90    // This should be:
91    // ```impl<F, I: From<F>> From<EncodedEvent<F>> for EncodedEvent<I>```
92    // however, the compiler rejects that due to conflicting
93    // implementations of `From` due to the generic
94    // ```impl<T> From<T> for T```
95    pub fn from<F>(that: EncodedEvent<F>) -> Self
96    where
97        I: From<F>,
98    {
99        Self {
100            item: I::from(that.item),
101            finalizers: that.finalizers,
102            byte_size: that.byte_size,
103            json_byte_size: that.json_byte_size,
104        }
105    }
106
107    /// Remap the item using an adapter
108    pub fn map<T>(self, doit: impl Fn(I) -> T) -> EncodedEvent<T> {
109        EncodedEvent {
110            item: doit(self.item),
111            finalizers: self.finalizers,
112            byte_size: self.byte_size,
113            json_byte_size: self.json_byte_size,
114        }
115    }
116}
117
118/// Joins namespace with name via delimiter if namespace is present.
119pub fn encode_namespace<'a>(
120    namespace: Option<&str>,
121    delimiter: char,
122    name: impl Into<Cow<'a, str>>,
123) -> String {
124    let name = name.into();
125    namespace
126        .map(|namespace| format!("{namespace}{delimiter}{name}"))
127        .unwrap_or_else(|| name.into_owned())
128}
129
130/// Marker trait for types that can hold a batch of events
131pub trait ElementCount {
132    fn element_count(&self) -> usize;
133}
134
135impl<T> ElementCount for Vec<T> {
136    fn element_count(&self) -> usize {
137        self.len()
138    }
139}
140
141pub fn timezone_to_offset(tz: TimeZone) -> Option<FixedOffset> {
142    match tz {
143        TimeZone::Local => Some(*Utc::now().with_timezone(&chrono::Local).offset()),
144        TimeZone::Named(tz) => Some(Utc::now().with_timezone(&tz).offset().fix()),
145    }
146}
vector/sinks/util/mod.rs

vector/sinks/util/
mod.rs
