oa_gateway/config/
stomp.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
//! `[stomp]` section: the STOMP client toward an ActiveMQ (or other) broker.
//!
//! Off when the section is omitted, so a config that never mentions it
//! does not try to connect. A present `[stomp]` table turns it on
//! unless `enabled = false`. Java CAL/OpenWire peers share the same
//! destinations when the topic names match.

use oa_gateway_stomp::Secret;
use serde::Deserialize;

use super::default_true;

/// STOMP adapter settings.
///
/// `broker` stays a string so a hostname can be resolved at startup.
/// Empty `login` and `passcode` mean omit those CONNECT headers, not send
/// blanks.
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct StompSection {
    /// Whether to spawn the adapter. `true` when the section is present.
    #[serde(default = "default_true")]
    pub(crate) enabled: bool,
    /// Engine adapter id. Defaults to `"stomp"`.
    #[serde(default = "default_stomp_id")]
    pub(crate) id: String,
    /// Broker address, `host:port`. Defaults to `127.0.0.1:61613`.
    #[serde(default = "default_stomp_broker")]
    pub(crate) broker: String,
    /// STOMP `host` header. ActiveMQ Classic typically wants `"/"`.
    #[serde(default = "default_stomp_host")]
    pub(crate) host: String,
    /// CONNECT login. Empty (the default) omits the header.
    #[serde(default)]
    pub(crate) login: String,
    /// CONNECT passcode. Empty (the default) omits the header. Held as a
    /// [`Secret`] so a `Debug` of the parsed config cannot print it.
    #[serde(default)]
    pub(crate) passcode: Secret,
    /// Prefix prepended to each topic to form a STOMP destination.
    /// Defaults to `"/topic/"`.
    #[serde(default = "default_stomp_prefix")]
    pub(crate) destination_prefix: String,
    /// Engine topic names, and the suffixes of the STOMP destinations
    /// bridged both ways. Defaults to `["demo"]`.
    #[serde(default = "default_stomp_topics")]
    pub(crate) topics: Vec<String>,
    /// Peel A-GRA Rx/Tx hex wrappers on inbound frames and fan out
    /// wrapper plus inner. Defaults to `true`.
    #[serde(default = "default_true")]
    pub(crate) unwrap_ma_payloads: bool,
    /// Retry the broker after a dropped session. Defaults to `true`.
    #[serde(default = "default_true")]
    pub(crate) reconnect: bool,
    /// Seconds to wait between reconnect attempts. Defaults to `1`.
    #[serde(default = "default_stomp_reconnect_delay_secs")]
    pub(crate) reconnect_delay_secs: u64,
    /// Seconds for TCP connect and for the CONNECTED wait, each.
    /// Defaults to `5`.
    #[serde(default = "default_stomp_connect_timeout_secs")]
    pub(crate) connect_timeout_secs: u64,
    /// Skip outbound SEND when the envelope came from this adapter.
    /// Defaults to `true`.
    #[serde(default = "default_true")]
    pub(crate) suppress_echo: bool,
    /// `abort` or `reconnect` when a session task panics. Defaults to
    /// `"abort"`.
    #[serde(default = "default_stomp_on_panic")]
    pub(crate) on_panic: String,
    /// Largest frame accepted from the broker, in bytes. Defaults to the
    /// adapter crate's limit.
    #[serde(default = "default_stomp_max_frame_size")]
    pub(crate) max_frame_size: usize,
    /// Wrap the broker connection in TLS. Defaults to `false`. ActiveMQ
    /// Classic's SSL transport connector conventionally listens on
    /// `61612`, not `61613`.
    #[serde(default)]
    pub(crate) tls: bool,
    /// PEM bundle of certificate authorities the broker's certificate
    /// must chain to. Empty (the default) uses the operating system
    /// trust store.
    #[serde(default)]
    pub(crate) tls_ca: String,
    /// Name checked against the broker's certificate. Empty (the
    /// default) uses the host part of `broker`.
    #[serde(default)]
    pub(crate) tls_server_name: String,
    /// PEM certificate chain presented to the broker, leaf certificate
    /// first. Empty (the default) presents nothing. Requires
    /// `tls_client_key` and `tls = true`.
    #[serde(default)]
    pub(crate) tls_client_cert: String,
    /// PEM private key for `tls_client_cert`, in PKCS#8, PKCS#1, or SEC1
    /// form. Empty (the default) presents nothing. Requires
    /// `tls_client_cert`.
    #[serde(default)]
    pub(crate) tls_client_key: String,
}

impl Default for StompSection {
    fn default() -> Self {
        Self {
            enabled: false,
            id: default_stomp_id(),
            broker: default_stomp_broker(),
            host: default_stomp_host(),
            login: String::new(),
            passcode: Secret::default(),
            destination_prefix: default_stomp_prefix(),
            topics: default_stomp_topics(),
            unwrap_ma_payloads: true,
            reconnect: true,
            reconnect_delay_secs: default_stomp_reconnect_delay_secs(),
            connect_timeout_secs: default_stomp_connect_timeout_secs(),
            suppress_echo: true,
            on_panic: default_stomp_on_panic(),
            max_frame_size: default_stomp_max_frame_size(),
            tls: false,
            tls_ca: String::new(),
            tls_server_name: String::new(),
            tls_client_cert: String::new(),
            tls_client_key: String::new(),
        }
    }
}

impl StompSection {
    /// Parses [`Self::on_panic`] into [`oa_gateway_stomp::OnPanic`].
    ///
    /// # Errors
    ///
    /// Returns an error if the string is not `abort` or `reconnect`.
    pub(crate) fn on_panic_mode(&self) -> Result<oa_gateway_stomp::OnPanic, String> {
        self.on_panic
            .parse()
            .map_err(|err| format!("stomp.on_panic: {err}"))
    }
}

fn default_stomp_id() -> String {
    "stomp".into()
}
fn default_stomp_broker() -> String {
    "127.0.0.1:61613".into()
}
fn default_stomp_host() -> String {
    "/".into()
}
fn default_stomp_prefix() -> String {
    "/topic/".into()
}
fn default_stomp_topics() -> Vec<String> {
    vec!["demo".into()]
}
fn default_stomp_max_frame_size() -> usize {
    oa_gateway_stomp::DEFAULT_MAX_FRAME_SIZE
}
fn default_stomp_reconnect_delay_secs() -> u64 {
    1
}
fn default_stomp_connect_timeout_secs() -> u64 {
    5
}
fn default_stomp_on_panic() -> String {
    oa_gateway_stomp::OnPanic::default().to_string()
}