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                        );
31                        None
32                    }
33                }
34            })
35            .into_driver(self.service)
36            .run()
37            .await
38    }
39}
40
41#[async_trait::async_trait]
42impl StreamSink<Event> for PostgresSink {
43    async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
44        self.run_inner(input).await
45    }
46}
vector/sinks/postgres/sink.rs

vector/sinks/postgres/
sink.rs
