vector/sinks/gcs_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 GcsSink<Svc, RB, P = KeyPartitioner> {
8    service: Svc,
9    request_builder: RB,
10    partitioner: P,
11    batcher_settings: BatcherSettings,
12    protocol: &'static str,
13}
14
15impl<Svc, RB, P> GcsSink<Svc, RB, P> {
16    pub const fn new(
17        service: Svc,
18        request_builder: RB,
19        partitioner: P,
20        batcher_settings: BatcherSettings,
21        protocol: &'static str,
22    ) -> Self {
23        Self {
24            service,
25            request_builder,
26            partitioner,
27            batcher_settings,
28            protocol,
29        }
30    }
31}
32
33impl<Svc, RB, P> GcsSink<Svc, RB, P>
34where
35    Svc: Service<RB::Request> + Send + 'static,
36    Svc::Future: Send + 'static,
37    Svc::Response: DriverResponse + Send + 'static,
38    Svc::Error: fmt::Debug + Into<crate::Error> + Send,
39    RB: RequestBuilder<(String, Vec<Event>)> + Send + Sync + 'static,
40    RB::Error: fmt::Display + Send,
41    RB::Request: Finalizable + MetaDescriptive + Send,
42    P: Partitioner<Item = Event, Key = Option<String>> + Unpin + Send,
43    P::Key: Eq + std::hash::Hash + Clone,
44    P::Item: ByteSizeOf,
45{
46    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
47        let partitioner = self.partitioner;
48        let settings = self.batcher_settings;
49
50        let request_builder = self.request_builder;
51
52        input
53            .batched_partitioned(partitioner, settings.timeout, |_| {
54                settings.as_byte_size_config()
55            })
56            .filter_map(|(key, batch)| async move {
57                // A `TemplateRenderingError` will have been emitted by `KeyPartitioner` if the key here is `None`,
58                // thus no further `EventsDropped` event needs emitting at this stage.
59                key.map(move |k| (k, batch))
60            })
61            .request_builder(default_request_builder_concurrency_limit(), request_builder)
62            .filter_map(|request| async move {
63                match request {
64                    Err(error) => {
65                        emit!(SinkRequestBuildError { error });
66                        None
67                    }
68                    Ok(req) => Some(req),
69                }
70            })
71            .into_driver(self.service)
72            .protocol(self.protocol)
73            .run()
74            .await
75    }
76}
77
78#[async_trait]
79impl<Svc, RB, P> StreamSink<Event> for GcsSink<Svc, RB, P>
80where
81    Svc: Service<RB::Request> + Send + 'static,
82    Svc::Future: Send + 'static,
83    Svc::Response: DriverResponse + Send + 'static,
84    Svc::Error: fmt::Debug + Into<crate::Error> + Send,
85    RB: RequestBuilder<(String, Vec<Event>)> + Send + Sync + 'static,
86    RB::Error: fmt::Display + Send,
87    RB::Request: Finalizable + MetaDescriptive + Send,
88    P: Partitioner<Item = Event, Key = Option<String>> + Unpin + Send,
89    P::Key: Eq + std::hash::Hash + Clone,
90    P::Item: ByteSizeOf,
91{
92    async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
93        self.run_inner(input).await
94    }
95}
vector/sinks/gcs_common/sink.rs

vector/sinks/gcs_common/
sink.rs
