vector/sinks/s3_common/
sink.rs

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

vector/sinks/s3_common/
sink.rs
