varpulis_connectors/
sink.rs

1//! Sink trait and error types for outputting processed events
2//!
3//! This module defines the core `Sink` trait that all event output destinations
4//! must implement, along with the `SinkError` error type.
5
6use std::sync::Arc;
7
8use async_trait::async_trait;
9use varpulis_core::Event;
10
11use crate::types::ConnectorError;
12
13/// Errors produced by sink operations.
14#[derive(Debug, thiserror::Error)]
15pub enum SinkError {
16    /// I/O error (file writes, network, etc.)
17    #[error("I/O error: {0}")]
18    Io(#[from] std::io::Error),
19
20    /// Serialization error (JSON encoding)
21    #[error("serialization error: {0}")]
22    Serialization(#[from] serde_json::Error),
23
24    /// HTTP request error
25    #[error("HTTP error: {0}")]
26    Http(#[from] reqwest::Error),
27
28    /// Connector-level error
29    #[error("connector error: {0}")]
30    Connector(#[from] ConnectorError),
31
32    /// Generic error with message
33    #[error("{0}")]
34    Other(String),
35}
36
37impl SinkError {
38    /// Create a generic error from a displayable value.
39    pub fn other(msg: impl std::fmt::Display) -> Self {
40        Self::Other(msg.to_string())
41    }
42}
43
44/// Trait for event sinks
45#[async_trait]
46pub trait Sink: Send + Sync {
47    /// Name of this sink
48    fn name(&self) -> &str;
49
50    /// Establish connection to the external system.
51    ///
52    /// Called once after sink creation to establish any necessary connections.
53    /// The default implementation is a no-op for sinks that connect eagerly.
54    async fn connect(&self) -> Result<(), SinkError> {
55        Ok(())
56    }
57
58    /// Send an event to this sink
59    async fn send(&self, event: &Event) -> Result<(), SinkError>;
60
61    /// Send a batch of events to this sink.
62    ///
63    /// Default implementation calls `send()` for each event.
64    /// Connectors should override this to amortize lock/syscall overhead.
65    async fn send_batch(&self, events: &[Arc<Event>]) -> Result<(), SinkError> {
66        for event in events {
67            self.send(event).await?;
68        }
69        Ok(())
70    }
71
72    /// Flush any buffered data
73    async fn flush(&self) -> Result<(), SinkError>;
74
75    /// Close the sink
76    async fn close(&self) -> Result<(), SinkError>;
77}
78
79/// Adapter: wraps a SinkConnector as a Sink for use in the sink registry.
80pub struct SinkConnectorAdapter {
81    name: String,
82    inner: tokio::sync::Mutex<Box<dyn crate::types::SinkConnector>>,
83}
84
85impl std::fmt::Debug for SinkConnectorAdapter {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct("SinkConnectorAdapter")
88            .finish_non_exhaustive()
89    }
90}
91
92impl SinkConnectorAdapter {
93    /// Create a new adapter wrapping a SinkConnector.
94    pub fn new(name: &str, connector: Box<dyn crate::types::SinkConnector>) -> Self {
95        Self {
96            name: name.to_string(),
97            inner: tokio::sync::Mutex::new(connector),
98        }
99    }
100}
101
102#[async_trait]
103impl Sink for SinkConnectorAdapter {
104    fn name(&self) -> &str {
105        &self.name
106    }
107    async fn connect(&self) -> Result<(), SinkError> {
108        let mut inner = self.inner.lock().await;
109        inner.connect().await.map_err(SinkError::from)
110    }
111    async fn send(&self, event: &Event) -> Result<(), SinkError> {
112        let inner = self.inner.lock().await;
113        inner.send(event).await.map_err(SinkError::from)
114    }
115    async fn send_batch(&self, events: &[Arc<Event>]) -> Result<(), SinkError> {
116        let inner = self.inner.lock().await;
117        for event in events {
118            inner.send(event).await.map_err(SinkError::from)?;
119        }
120        Ok(())
121    }
122    async fn flush(&self) -> Result<(), SinkError> {
123        let inner = self.inner.lock().await;
124        inner.flush().await.map_err(SinkError::from)
125    }
126    async fn close(&self) -> Result<(), SinkError> {
127        let inner = self.inner.lock().await;
128        inner.close().await.map_err(SinkError::from)
129    }
130}
varpulis_connectors/sink.rs

varpulis_connectors/
sink.rs
