faucet_sink_stdout/
config.rs

1//! Stdout/stderr sink configuration.
2
3use faucet_core::DEFAULT_BATCH_SIZE;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// Which standard stream to write records to.
8#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
9#[serde(rename_all = "lowercase")]
10pub enum StdStream {
11    /// Standard output (default). Honors shell redirection.
12    #[default]
13    Stdout,
14    /// Standard error. Useful when stdout is reserved for piping pipeline output.
15    Stderr,
16}
17
18/// How each record should be serialized before writing.
19#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
20#[serde(rename_all = "snake_case")]
21pub enum StdoutFormat {
22    /// One compact JSON object per line (default — matches JSONL format).
23    #[default]
24    JsonLines,
25    /// Indented JSON, separated by newlines. Easier to read but not a single-line format.
26    PrettyJson,
27    /// Tab-separated values, with each record's keys sorted alphabetically.
28    /// Scalars are emitted as-is; nested objects/arrays are emitted as compact JSON.
29    Tsv,
30    /// RFC-4180 comma-separated values, with each record's keys sorted
31    /// alphabetically. Same column-resolution as [`StdoutFormat::Tsv`] (one
32    /// line per record, keys sorted, no header row); values containing commas,
33    /// quotes, or newlines are properly quoted. Nested objects/arrays are
34    /// emitted as compact JSON.
35    Csv,
36}
37
38/// Configuration for the stdout/stderr sink.
39#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
40pub struct StdoutSinkConfig {
41    /// Which standard stream to write to.
42    #[serde(default)]
43    pub destination: StdStream,
44    /// Output format.
45    #[serde(default)]
46    pub format: StdoutFormat,
47    /// Flush the underlying writer after every record instead of at batch boundaries.
48    /// Tradeoff: lower latency for live preview, slightly lower throughput.
49    #[serde(default)]
50    pub flush_per_record: bool,
51    /// Stop writing after this many records have been emitted. Subsequent
52    /// `write_batch` calls become no-ops. `None` means unlimited.
53    #[serde(default)]
54    pub max_records: Option<usize>,
55    /// Records per upstream [`StreamPage`](faucet_core::StreamPage). The
56    /// stdout sink writes records to the chosen standard stream one at a time
57    /// through a buffered writer, so this field has **no behavioural impact**
58    /// at the sink — it is exposed purely for config parity across every sink
59    /// in the workspace. Defaults to [`DEFAULT_BATCH_SIZE`].
60    ///
61    /// `batch_size = 0` (the "no batching" sentinel) and any positive value
62    /// produce byte-for-byte identical output for this sink: each record is
63    /// serialised and written individually regardless of how upstream chunked
64    /// the page.
65    #[serde(default = "default_batch_size")]
66    pub batch_size: usize,
67}
68
69fn default_batch_size() -> usize {
70    DEFAULT_BATCH_SIZE
71}
72
73impl Default for StdoutSinkConfig {
74    fn default() -> Self {
75        Self {
76            destination: StdStream::default(),
77            format: StdoutFormat::default(),
78            flush_per_record: false,
79            max_records: None,
80            batch_size: DEFAULT_BATCH_SIZE,
81        }
82    }
83}
84
85impl StdoutSinkConfig {
86    /// Create a new config with all defaults (stdout, JSON Lines, no limit).
87    pub fn new() -> Self {
88        Self::default()
89    }
90
91    /// Send records to the given standard stream.
92    pub fn destination(mut self, destination: StdStream) -> Self {
93        self.destination = destination;
94        self
95    }
96
97    /// Choose the output format.
98    pub fn format(mut self, format: StdoutFormat) -> Self {
99        self.format = format;
100        self
101    }
102
103    /// Flush after every record.
104    pub fn flush_per_record(mut self, flush_per_record: bool) -> Self {
105        self.flush_per_record = flush_per_record;
106        self
107    }
108
109    /// Stop after writing `n` records total.
110    pub fn max_records(mut self, max_records: usize) -> Self {
111        self.max_records = Some(max_records);
112        self
113    }
114
115    /// Set the per-page record count hint reported alongside other sink
116    /// configs.
117    ///
118    /// This sink writes per-record through a buffered writer, so the value is
119    /// observably a no-op: `0` (the "no batching" sentinel) and any positive
120    /// value produce the same stdout/stderr output. Present for symmetry with
121    /// sinks whose `batch_size` does drive I/O sizing (e.g. SQL multi-row
122    /// inserts, BigQuery streaming inserts).
123    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
124        self.batch_size = batch_size;
125        self
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    #[test]
134    fn defaults() {
135        let c = StdoutSinkConfig::new();
136        assert_eq!(c.destination, StdStream::Stdout);
137        assert_eq!(c.format, StdoutFormat::JsonLines);
138        assert!(!c.flush_per_record);
139        assert!(c.max_records.is_none());
140    }
141
142    #[test]
143    fn builder_chains() {
144        let c = StdoutSinkConfig::new()
145            .destination(StdStream::Stderr)
146            .format(StdoutFormat::PrettyJson)
147            .flush_per_record(true)
148            .max_records(10);
149        assert_eq!(c.destination, StdStream::Stderr);
150        assert_eq!(c.format, StdoutFormat::PrettyJson);
151        assert!(c.flush_per_record);
152        assert_eq!(c.max_records, Some(10));
153    }
154
155    #[test]
156    fn serde_round_trip() {
157        let c = StdoutSinkConfig::new()
158            .destination(StdStream::Stderr)
159            .format(StdoutFormat::Tsv);
160        let json = serde_json::to_string(&c).unwrap();
161        let back: StdoutSinkConfig = serde_json::from_str(&json).unwrap();
162        assert_eq!(back.destination, StdStream::Stderr);
163        assert_eq!(back.format, StdoutFormat::Tsv);
164    }
165
166    #[test]
167    fn deserialize_from_minimal_json() {
168        let c: StdoutSinkConfig = serde_json::from_str("{}").unwrap();
169        assert_eq!(c.destination, StdStream::Stdout);
170        assert_eq!(c.format, StdoutFormat::JsonLines);
171    }
172
173    #[test]
174    fn batch_size_defaults_to_default_batch_size() {
175        let c = StdoutSinkConfig::new();
176        assert_eq!(c.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
177    }
178
179    #[test]
180    fn with_batch_size_overrides_default() {
181        let c = StdoutSinkConfig::new().with_batch_size(250);
182        assert_eq!(c.batch_size, 250);
183    }
184
185    #[test]
186    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
187        let c = StdoutSinkConfig::new().with_batch_size(0);
188        assert_eq!(c.batch_size, 0);
189        assert!(faucet_core::validate_batch_size(c.batch_size).is_ok());
190    }
191
192    #[test]
193    fn batch_size_above_max_is_rejected_by_validate_batch_size() {
194        let c = StdoutSinkConfig::new().with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
195        assert!(faucet_core::validate_batch_size(c.batch_size).is_err());
196    }
197
198    #[test]
199    fn batch_size_deserializes_from_json() {
200        let json = r#"{
201            "destination": "stdout",
202            "format": "json_lines",
203            "batch_size": 500
204        }"#;
205        let c: StdoutSinkConfig = serde_json::from_str(json).unwrap();
206        assert_eq!(c.batch_size, 500);
207    }
208
209    #[test]
210    fn batch_size_defaults_when_missing_in_json() {
211        let c: StdoutSinkConfig = serde_json::from_str("{}").unwrap();
212        assert_eq!(c.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
213    }
214}
faucet_sink_stdout/config.rs

faucet_sink_stdout/
config.rs
