vector/sinks/azure_common/
sink.rs

1use std::fmt;
2
3use vector_lib::{event::Event, partition::Partitioner};
4
5use crate::sinks::{prelude::*, util::partitioner::KeyPartitioner};
6
7pub struct AzureBlobSink<Svc, RB, P = KeyPartitioner> {
8    service: Svc,
9    request_builder: RB,
10    partitioner: P,
11    batcher_settings: BatcherSettings,
12}
13
14impl<Svc, RB, P> AzureBlobSink<Svc, RB, P> {
15    pub const fn new(
16        service: Svc,
17        request_builder: RB,
18        partitioner: P,
19        batcher_settings: BatcherSettings,
20    ) -> Self {
21        Self {
22            service,
23            request_builder,
24            partitioner,
25            batcher_settings,
26        }
27    }
28}
29
30impl<Svc, RB, P> AzureBlobSink<Svc, RB, P>
31where
32    Svc: Service<RB::Request> + Send + 'static,
33    Svc::Future: Send + 'static,
34    Svc::Response: DriverResponse + Send + 'static,
35    Svc::Error: fmt::Debug + Into<crate::Error> + Send,
36    RB: RequestBuilder<(String, Vec<Event>)> + Send + Sync + 'static,
37    RB::Error: fmt::Display + Send,
38    RB::Request: Finalizable + MetaDescriptive + Send,
39    P: Partitioner<Item = Event, Key = Option<String>> + Unpin + Send,
40    P::Key: Eq + std::hash::Hash + Clone,
41    P::Item: ByteSizeOf,
42{
43    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
44        let partitioner = self.partitioner;
45        let settings = self.batcher_settings;
46
47        let request_builder = self.request_builder;
48
49        input
50            .batched_partitioned(partitioner, || settings.as_byte_size_config())
51            .filter_map(|(key, batch)| async move {
52                // We don't need to emit an error here if the event is dropped since this will occur if the template
53                // couldn't be rendered during the partitioning. A `TemplateRenderingError` is already emitted when
54                // that occurs.
55                key.map(move |k| (k, batch))
56            })
57            .request_builder(default_request_builder_concurrency_limit(), request_builder)
58            .filter_map(|request| async move {
59                match request {
60                    Err(error) => {
61                        emit!(SinkRequestBuildError { error });
62                        None
63                    }
64                    Ok(req) => Some(req),
65                }
66            })
67            .into_driver(self.service)
68            .protocol("https")
69            .run()
70            .await
71    }
72}
73
74#[async_trait]
75impl<Svc, RB, P> StreamSink<Event> for AzureBlobSink<Svc, RB, P>
76where
77    Svc: Service<RB::Request> + Send + 'static,
78    Svc::Future: Send + 'static,
79    Svc::Response: DriverResponse + Send + 'static,
80    Svc::Error: fmt::Debug + Into<crate::Error> + Send,
81    RB: RequestBuilder<(String, Vec<Event>)> + Send + Sync + 'static,
82    RB::Error: fmt::Display + Send,
83    RB::Request: Finalizable + MetaDescriptive + Send,
84    P: Partitioner<Item = Event, Key = Option<String>> + Unpin + Send,
85    P::Key: Eq + std::hash::Hash + Clone,
86    P::Item: ByteSizeOf,
87{
88    async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
89        self.run_inner(input).await
90    }
91}
vector/sinks/azure_common/sink.rs

vector/sinks/azure_common/
sink.rs
