faucet_sink_sftp/
config.rs

1//! SFTP sink configuration.
2
3use faucet_common_sftp::SftpConnectionConfig;
4use faucet_core::DEFAULT_BATCH_SIZE;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// Configuration for the SFTP sink connector.
9#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
10pub struct SftpSinkConfig {
11    /// Shared SFTP connection settings (host, port, username, auth, host-key
12    /// policy). Flattened, so its fields sit at the top level of the config.
13    #[serde(flatten)]
14    pub connection: SftpConnectionConfig,
15    /// Remote directory prefix under which JSON Lines objects are written.
16    pub path: String,
17    /// File extension for written objects (default: `.jsonl`).
18    #[serde(default = "default_file_extension")]
19    pub file_extension: String,
20    /// Records per written object. When a `write_batch` call hands the sink
21    /// `N` records with `batch_size = M > 0`, the sink writes `ceil(N / M)`
22    /// objects. `batch_size = 0` writes whatever upstream hands it as a single
23    /// object. Defaults to [`DEFAULT_BATCH_SIZE`].
24    #[serde(default = "default_batch_size")]
25    pub batch_size: usize,
26}
27
28fn default_file_extension() -> String {
29    ".jsonl".to_string()
30}
31
32fn default_batch_size() -> usize {
33    DEFAULT_BATCH_SIZE
34}
35
36impl SftpSinkConfig {
37    /// Build a sink config from a connection and a remote directory prefix.
38    pub fn new(connection: SftpConnectionConfig, path: impl Into<String>) -> Self {
39        Self {
40            connection,
41            path: path.into(),
42            file_extension: default_file_extension(),
43            batch_size: DEFAULT_BATCH_SIZE,
44        }
45    }
46
47    /// Set the file extension for written objects.
48    pub fn file_extension(mut self, ext: impl Into<String>) -> Self {
49        self.file_extension = ext.into();
50        self
51    }
52
53    /// Set the per-object record count.
54    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
55        self.batch_size = batch_size;
56        self
57    }
58}
59
60#[cfg(test)]
61mod tests {
62    use super::*;
63    use faucet_common_sftp::SftpConnectionConfig;
64
65    fn conn() -> SftpConnectionConfig {
66        SftpConnectionConfig::with_password("h", "u", "p")
67    }
68
69    #[test]
70    fn defaults() {
71        let cfg = SftpSinkConfig::new(conn(), "/out");
72        assert_eq!(cfg.path, "/out");
73        assert_eq!(cfg.file_extension, ".jsonl");
74        assert_eq!(cfg.batch_size, DEFAULT_BATCH_SIZE);
75    }
76
77    #[test]
78    fn deserializes_flat_shape() {
79        let json = r#"{
80            "host": "sftp.example.com",
81            "username": "user",
82            "type": "password",
83            "config": { "password": "secret" },
84            "path": "/upload",
85            "batch_size": 0
86        }"#;
87        let cfg: SftpSinkConfig = serde_json::from_str(json).unwrap();
88        assert_eq!(cfg.connection.host, "sftp.example.com");
89        assert_eq!(cfg.path, "/upload");
90        assert_eq!(cfg.batch_size, 0);
91        assert_eq!(cfg.file_extension, ".jsonl");
92    }
93
94    #[test]
95    fn batch_size_zero_is_valid_sentinel() {
96        let cfg = SftpSinkConfig::new(conn(), "/o").with_batch_size(0);
97        assert!(faucet_core::validate_batch_size(cfg.batch_size).is_ok());
98    }
99}
faucet_sink_sftp/config.rs

faucet_sink_sftp/
config.rs
