vector/sinks/postgres/
sink.rs

1use super::service::{PostgresRequest, PostgresRetryLogic, PostgresService};
2use crate::sinks::prelude::*;
3
4pub struct PostgresSink {
5    service: Svc<PostgresService, PostgresRetryLogic>,
6    batch_settings: BatcherSettings,
7}
8
9impl PostgresSink {
10    pub const fn new(
11        service: Svc<PostgresService, PostgresRetryLogic>,
12        batch_settings: BatcherSettings,
13    ) -> Self {
14        Self {
15            service,
16            batch_settings,
17        }
18    }
19
20    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
21        input
22            .batched(self.batch_settings.as_byte_size_config())
23            .filter_map(|events| async move {
24                match PostgresRequest::try_from(events) {
25                    Ok(request) => Some(request),
26                    Err(e) => {
27                        warn!(
28                            message = "Error creating postgres sink's request.",
29                            error = %e,
30                            internal_log_rate_limit=true
31                        );
32                        None
33                    }
34                }
35            })
36            .into_driver(self.service)
37            .run()
38            .await
39    }
40}
41
42#[async_trait::async_trait]
43impl StreamSink<Event> for PostgresSink {
44    async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
45        self.run_inner(input).await
46    }
47}
vector/sinks/postgres/sink.rs

vector/sinks/postgres/
sink.rs
