vector/sinks/s3_common/
sink.rs1use std::{fmt, hash::Hash};
2
3use crate::sinks::prelude::*;
4
5use super::partitioner::S3PartitionKey;
6use vector_lib::{event::Event, partition::Partitioner};
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.as_byte_size_config())
51 .filter_map(|(key, batch)| async move { key.map(move |k| (k, batch)) })
52 .request_builder(default_request_builder_concurrency_limit(), request_builder)
53 .filter_map(|request| async move {
54 match request {
55 Err(error) => {
56 emit!(SinkRequestBuildError { error });
57 None
58 }
59 Ok(req) => Some(req),
60 }
61 })
62 .into_driver(self.service)
63 .run()
64 .await
65 }
66}
67
68#[async_trait]
69impl<Svc, RB, P> StreamSink<Event> for S3Sink<Svc, RB, P>
70where
71 Svc: Service<RB::Request> + Send + 'static,
72 Svc::Future: Send + 'static,
73 Svc::Response: DriverResponse + Send + 'static,
74 Svc::Error: fmt::Debug + Into<crate::Error> + Send,
75 RB: RequestBuilder<(S3PartitionKey, Vec<Event>)> + Send + Sync + 'static,
76 RB::Error: fmt::Display + Send,
77 RB::Request: Finalizable + MetaDescriptive + Send,
78 P: Partitioner<Item = Event, Key = Option<S3PartitionKey>> + Unpin + Send,
79 P::Key: Eq + Hash + Clone,
80 P::Item: ByteSizeOf,
81{
82 async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
83 self.run_inner(input).await
84 }
85}