vector/sinks/gcs_common/
sink.rs

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

vector/sinks/gcs_common/
sink.rs
