vector/sinks/azure_common/
sink.rs

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

vector/sinks/azure_common/
sink.rs
