kernel/
port_entry.rs

1//! One configured port: which adapter (`type`) and its options.
2//!
3//! `[source]`, `[stream]`, and every `[sinks.<name>]` table in `flusso.toml`
4//! is a [`PortEntry`]. The config layer reads the `type` and carries the rest
5//! as an opaque [`Options`] map; the composition root looks the kind up in
6//! its adapter registry and hands the options to that adapter's config type.
7//! Serialized as the file writes it: `type = "…"` plus the options flattened
8//! beside it, keys sorted, so the lock is byte-stable.
9//!
10//! ```
11//! use kernel::PortEntry;
12//!
13//! let entry: PortEntry = toml::from_str(r#"
14//!     type = "opensearch"
15//!     url = "https://search:9200"
16//!     batch_size = 500
17//! "#).unwrap();
18//! assert_eq!(entry.kind, "opensearch");
19//! assert_eq!(entry.options.get("batch_size").and_then(|v| v.as_i64()), Some(500));
20//! assert_eq!(toml::to_string(&entry).unwrap(), "type = \"opensearch\"\nbatch_size = 500\nurl = \"https://search:9200\"\n");
21//! ```
22
23use serde::{Deserialize, Serialize};
24
25use crate::options::Options;
26
27/// One port entry of a deployment: the adapter kind and its options.
28#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
29pub struct PortEntry {
30    /// The adapter that implements the port, as written in `type = "…"`.
31    #[serde(rename = "type")]
32    pub kind: String,
33    /// Everything else under the table, uninterpreted here.
34    #[serde(flatten)]
35    pub options: Options,
36}
37
38impl PortEntry {
39    /// An entry of `kind` with no options, so every adapter default applies.
40    pub fn new(kind: impl Into<String>) -> Self {
41        Self {
42            kind: kind.into(),
43            options: Options::empty(),
44        }
45    }
46
47    /// An entry of `kind` with the given options.
48    pub fn with_options(kind: impl Into<String>, options: Options) -> Self {
49        Self {
50            kind: kind.into(),
51            options,
52        }
53    }
54}
55
56#[cfg(test)]
57#[allow(clippy::unwrap_used)]
58mod tests {
59    use super::*;
60
61    #[test]
62    fn type_alone_is_an_empty_options_map() {
63        let entry: PortEntry = toml::from_str("type = \"stdout\"\n").unwrap();
64        assert_eq!(entry, PortEntry::new("stdout"));
65        assert!(entry.options.is_empty());
66    }
67
68    #[test]
69    fn missing_type_is_an_error() {
70        let error = toml::from_str::<PortEntry>("pretty = true\n").unwrap_err();
71        assert!(error.to_string().contains("type"), "{error}");
72    }
73
74    #[test]
75    fn nested_tables_round_trip() {
76        let text = "type = \"postgres\"\n\n[connection_url]\nhost = \"db\"\nport = 5432\n";
77        let entry: PortEntry = toml::from_str(text).unwrap();
78        let again = toml::to_string(&entry).unwrap();
79        assert_eq!(again, text);
80    }
81}
82
83/// The uninterpreted shape: a `type` plus any table. The composition root
84/// replaces this with one alternative per registered adapter when it assembles
85/// the editor schema.
86impl schemars::JsonSchema for PortEntry {
87    fn schema_name() -> std::borrow::Cow<'static, str> {
88        std::borrow::Cow::Borrowed("PortEntry")
89    }
90
91    fn json_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
92        schemars::json_schema!({
93            "type": "object",
94            "required": ["type"],
95            "properties": {
96                "type": { "type": "string", "description": "The adapter that implements this port." }
97            },
98            "additionalProperties": true
99        })
100    }
101}
kernel/port_entry.rs

kernel/
port_entry.rs
