simulator_api/
subscribe_config.rs

1//! Cross-cutting subscribe options shared by the client and server.
2//!
3//! In a subscribe request's positional `[filter, config]` params, the `config`
4//! object carries per-kind fields (e.g. `commitment`) plus these options. The
5//! client builds [`SubscribeConfig`] and merges it in; the server parses it back
6//! out — one typed home for the wire field names. Absent fields default off, so
7//! peers without support interoperate.
8
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11
12/// Compression negotiated for notification frames. `lowercase` so the variant
13/// matches the on-wire token (`"zstd"`).
14#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "lowercase")]
16pub enum Compression {
17    Zstd,
18}
19
20/// Cross-cutting subscribe options the client sets in the config object.
21#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct SubscribeConfig {
23    /// Last slot the client already processed. On reconnect the server replays
24    /// from this slot (inclusive) rather than streaming the full history.
25    #[serde(
26        rename = "replayFromSlot",
27        default,
28        skip_serializing_if = "Option::is_none"
29    )]
30    pub replay_from_slot: Option<i64>,
31    /// Notification-frame compression the client opted into. `None` keeps the
32    /// uncompressed `Text` path.
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub compression: Option<Compression>,
35    /// One account notification per write. Absent/false is Agave parity: one per
36    /// account per slot, carrying the slot's final state. Only `accountSubscribe`
37    /// and `programSubscribe` read it.
38    #[serde(
39        rename = "intraSlotUpdates",
40        default,
41        skip_serializing_if = "core::ops::Not::not"
42    )]
43    pub intra_slot_updates: bool,
44}
45
46impl SubscribeConfig {
47    /// Parse the cross-cutting options out of a subscribe config object. Per-kind
48    /// fields are ignored, and a missing or malformed object yields defaults
49    /// (all off), so a subscribe never fails on these optional fields.
50    pub fn from_value(config: Option<&Value>) -> Self {
51        config
52            .cloned()
53            .and_then(|v| serde_json::from_value(v).ok())
54            .unwrap_or_default()
55    }
56
57    /// Merge the set options into a subscribe request's config object (the second
58    /// positional param), leaving its per-kind fields untouched. A no-op if the
59    /// params don't have a config object.
60    pub fn apply_to(&self, params: &mut Value) {
61        let Some(config) = params.get_mut(1).and_then(Value::as_object_mut) else {
62            return;
63        };
64        if let Ok(Value::Object(fields)) = serde_json::to_value(self) {
65            config.extend(fields);
66        }
67    }
68}
69
70#[cfg(test)]
71mod tests {
72    use super::*;
73
74    #[test]
75    fn parses_set_options_and_ignores_per_kind_fields() {
76        let cfg = SubscribeConfig::from_value(Some(&serde_json::json!({
77            "commitment": "confirmed",
78            "replayFromSlot": 42,
79            "compression": "zstd",
80            "intraSlotUpdates": true,
81        })));
82        assert_eq!(cfg.replay_from_slot, Some(42));
83        assert_eq!(cfg.compression, Some(Compression::Zstd));
84        assert!(cfg.intra_slot_updates);
85    }
86
87    #[test]
88    fn absent_or_malformed_options_default_off() {
89        assert_eq!(
90            SubscribeConfig::from_value(None),
91            SubscribeConfig::default()
92        );
93        let cfg =
94            SubscribeConfig::from_value(Some(&serde_json::json!({ "commitment": "confirmed" })));
95        assert_eq!(cfg, SubscribeConfig::default());
96        assert!(!cfg.intra_slot_updates);
97        // An unrecognized compression token leaves compression off.
98        let cfg = SubscribeConfig::from_value(Some(&serde_json::json!({ "compression": "gzip" })));
99        assert_eq!(cfg.compression, None);
100    }
101
102    #[test]
103    fn apply_to_merges_into_config_without_clobbering() {
104        let mut params =
105            serde_json::json!([{ "mentions": ["prog"] }, { "commitment": "confirmed" }]);
106        SubscribeConfig {
107            replay_from_slot: Some(7),
108            compression: Some(Compression::Zstd),
109            intra_slot_updates: true,
110        }
111        .apply_to(&mut params);
112        assert_eq!(params[1]["commitment"], "confirmed");
113        assert_eq!(params[1]["replayFromSlot"], 7);
114        assert_eq!(params[1]["compression"], "zstd");
115        assert_eq!(params[1]["intraSlotUpdates"], true);
116    }
117
118    #[test]
119    fn apply_to_omits_unset_options() {
120        let mut params = serde_json::json!([{ "mentions": ["prog"] }, {}]);
121        SubscribeConfig::default().apply_to(&mut params);
122        assert_eq!(params[1].as_object().unwrap().len(), 0);
123    }
124}
simulator_api/subscribe_config.rs

simulator_api/
subscribe_config.rs
