vector/sinks/gcp_chronicle/
sink.rs1use std::fmt;
2
3use crate::sinks::gcp_chronicle::partitioner::{ChroniclePartitionKey, ChroniclePartitioner};
4use crate::sinks::prelude::*;
5
6pub struct ChronicleSink<Svc, RB> {
7 service: Svc,
8 request_builder: RB,
9 partitioner: ChroniclePartitioner,
10 batcher_settings: BatcherSettings,
11 protocol: &'static str,
12}
13
14impl<Svc, RB> ChronicleSink<Svc, RB> {
15 pub const fn new(
16 service: Svc,
17 request_builder: RB,
18 partitioner: ChroniclePartitioner,
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> ChronicleSink<Svc, RB>
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<(ChroniclePartitionKey, Vec<Event>)> + Send + Sync + 'static,
39 RB::Error: fmt::Display + Send,
40 RB::Request: Finalizable + MetaDescriptive + Send,
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 key.map(move |k| (k, batch))
54 })
55 .request_builder(default_request_builder_concurrency_limit(), request_builder)
56 .filter_map(|request| async move {
57 match request {
58 Err(error) => {
59 emit!(SinkRequestBuildError { error });
60 None
61 }
62 Ok(req) => Some(req),
63 }
64 })
65 .into_driver(self.service)
66 .protocol(self.protocol)
67 .run()
68 .await
69 }
70}
71
72#[async_trait]
73impl<Svc, RB> StreamSink<Event> for ChronicleSink<Svc, RB>
74where
75 Svc: Service<RB::Request> + Send + 'static,
76 Svc::Future: Send + 'static,
77 Svc::Response: DriverResponse + Send + 'static,
78 Svc::Error: fmt::Debug + Into<crate::Error> + Send,
79 RB: RequestBuilder<(ChroniclePartitionKey, Vec<Event>)> + Send + Sync + 'static,
80 RB::Error: fmt::Display + Send,
81 RB::Request: Finalizable + MetaDescriptive + Send,
82{
83 async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
84 self.run_inner(input).await
85 }
86}