rskit_config/sink/
memory.rs

1use std::collections::BTreeMap;
2use std::sync::Arc;
3
4use parking_lot::Mutex;
5use rskit_errors::AppResult;
6use rskit_util::SecretString;
7
8use super::ConfigSink;
9
10/// Shared mutable state behind an [`InMemoryConfigSink`].
11#[derive(Debug, Default)]
12struct State {
13    values: BTreeMap<String, SecretString>,
14}
15
16/// In-memory writable config store.
17///
18/// A process-local reference [`ConfigSink`] backed by a `BTreeMap`. Useful for tests, defaults,
19/// and composing override layers without touching disk or a remote backend. Cheaply cloneable;
20/// clones share the same underlying store.
21///
22/// With the `watch` feature enabled it also implements
23#[cfg_attr(
24    feature = "watch",
25    doc = " [`ConfigWatch`](crate::ConfigWatch): every [`set`](ConfigSink::set) and"
26)]
27#[cfg_attr(
28    not(feature = "watch"),
29    doc = " `ConfigWatch`: every [`set`](ConfigSink::set) and"
30)]
31/// [`remove`](ConfigSink::remove) fans a change event out to active subscribers.
32#[derive(Debug, Clone, Default)]
33pub struct InMemoryConfigSink {
34    state: Arc<Mutex<State>>,
35    #[cfg(feature = "watch")]
36    broadcaster: rskit_stream::Broadcaster<crate::watch::ConfigChange>,
37}
38
39impl InMemoryConfigSink {
40    /// Create an empty in-memory sink.
41    #[must_use]
42    pub fn new() -> Self {
43        Self::default()
44    }
45
46    /// Return the masked value stored at `key`, if present.
47    ///
48    /// The returned [`SecretString`] masks its plaintext in display, debug, and serialization;
49    /// use [`SecretString::expose`] to read the plaintext intentionally.
50    #[must_use]
51    pub fn get(&self, key: &str) -> Option<SecretString> {
52        self.state.lock().values.get(key).cloned()
53    }
54
55    /// Return the number of stored keys.
56    #[must_use]
57    pub fn len(&self) -> usize {
58        self.state.lock().values.len()
59    }
60
61    /// Return `true` if the store holds no keys.
62    #[must_use]
63    pub fn is_empty(&self) -> bool {
64        self.state.lock().values.is_empty()
65    }
66}
67
68impl ConfigSink for InMemoryConfigSink {
69    fn set(&self, key: &str, value: SecretString) -> AppResult<()> {
70        self.state.lock().values.insert(key.to_string(), value);
71        #[cfg(feature = "watch")]
72        self.broadcaster
73            .broadcast(&crate::watch::ConfigChange::Set {
74                key: key.to_string(),
75            });
76        Ok(())
77    }
78
79    fn remove(&self, key: &str) -> AppResult<()> {
80        let removed = self.state.lock().values.remove(key).is_some();
81        // Only emit a change event when state actually changed,
82        // so a no-op remove of an absent key never triggers a spurious reload for watchers.
83        #[cfg(feature = "watch")]
84        if removed {
85            self.broadcaster
86                .broadcast(&crate::watch::ConfigChange::Removed {
87                    key: key.to_string(),
88                });
89        }
90        #[cfg(not(feature = "watch"))]
91        let _ = removed;
92        Ok(())
93    }
94}
95
96#[cfg(feature = "watch")]
97impl crate::watch::ConfigWatch for InMemoryConfigSink {
98    fn watch(
99        &self,
100        cancel: tokio_util::sync::CancellationToken,
101    ) -> AppResult<crate::watch::ConfigChangeStream> {
102        Ok(self.broadcaster.subscribe(cancel))
103    }
104}
105
106#[cfg(test)]
107mod tests {
108    use super::*;
109
110    #[test]
111    fn set_then_get_round_trips_plaintext() {
112        let sink = InMemoryConfigSink::new();
113        sink.set("api_token", SecretString::new("s3cret")).unwrap();
114        assert_eq!(sink.get("api_token").unwrap().expose(), "s3cret");
115        assert_eq!(sink.len(), 1);
116    }
117
118    #[test]
119    fn set_replaces_existing_value() {
120        let sink = InMemoryConfigSink::new();
121        sink.set("k", SecretString::new("old")).unwrap();
122        sink.set("k", SecretString::new("new")).unwrap();
123        assert_eq!(sink.get("k").unwrap().expose(), "new");
124        assert_eq!(sink.len(), 1);
125    }
126
127    #[test]
128    fn remove_deletes_key_and_is_idempotent() {
129        let sink = InMemoryConfigSink::new();
130        sink.set("k", SecretString::new("v")).unwrap();
131        sink.remove("k").unwrap();
132        assert!(sink.get("k").is_none());
133        sink.remove("k").unwrap();
134        assert!(sink.is_empty());
135    }
136
137    #[test]
138    fn set_many_applies_all_entries() {
139        let sink = InMemoryConfigSink::new();
140        sink.set_many(vec![
141            ("a".to_string(), SecretString::new("1")),
142            ("b".to_string(), SecretString::new("2")),
143        ])
144        .unwrap();
145        assert_eq!(sink.get("a").unwrap().expose(), "1");
146        assert_eq!(sink.get("b").unwrap().expose(), "2");
147    }
148
149    #[test]
150    fn debug_does_not_leak_secret() {
151        let sink = InMemoryConfigSink::new();
152        sink.set("k", SecretString::new("topsecret")).unwrap();
153        let rendered = format!("{sink:?}");
154        assert!(!rendered.contains("topsecret"));
155    }
156
157    #[test]
158    fn clones_share_state() {
159        let sink = InMemoryConfigSink::new();
160        let clone = sink.clone();
161        sink.set("k", SecretString::new("v")).unwrap();
162        assert_eq!(clone.get("k").unwrap().expose(), "v");
163    }
164}
rskit_config/sink/memory.rs

rskit_config/sink/
memory.rs
