vector/sinks/postgres/
sink.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
use super::service::{PostgresRequest, PostgresRetryLogic, PostgresService};
use crate::sinks::prelude::*;

pub struct PostgresSink {
    service: Svc<PostgresService, PostgresRetryLogic>,
    batch_settings: BatcherSettings,
}

impl PostgresSink {
    pub const fn new(
        service: Svc<PostgresService, PostgresRetryLogic>,
        batch_settings: BatcherSettings,
    ) -> Self {
        Self {
            service,
            batch_settings,
        }
    }

    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
        input
            .batched(self.batch_settings.as_byte_size_config())
            .filter_map(|events| async move {
                match PostgresRequest::try_from(events) {
                    Ok(request) => Some(request),
                    Err(e) => {
                        warn!(
                            message = "Error creating postgres sink's request.",
                            error = %e,
                            internal_log_rate_limit=true
                        );
                        None
                    }
                }
            })
            .into_driver(self.service)
            .run()
            .await
    }
}

#[async_trait::async_trait]
impl StreamSink<Event> for PostgresSink {
    async fn run(mut self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
        self.run_inner(input).await
    }
}