road-runner-common 0.9.0

Shared Rust utilities for exchange ecosystem backend services.
Documentation 
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

//! Connection + security settings — the ONLY env-driven Kafka config in the
//! platform. Topics, consumer groups, `client.id` and delivery semantics live in
//! application code (see `KAFKA-CONVENTIONS.md`).

/// A single rdkafka client setting `(key, value)`.
///
/// This crate deliberately returns **plain settings**, never a version-specific
/// `rdkafka::ClientConfig`: the platform is spread across rdkafka 0.36–0.39, so a
/// typed config could not cross the crate boundary. Each service applies these to
/// its own `ClientConfig` (any rdkafka version):
///
/// ```ignore
/// let mut cfg = rdkafka::ClientConfig::new();
/// for (k, v) in consumer_settings(&settings, "cex-history", OffsetReset::Earliest) {
///     cfg.set(k, v);
/// }
/// ```
pub type Setting = (&'static str, String);

/// Standard Kafka connection + security, sourced from environment variables:
///
/// | Env | Meaning |
/// |---|---|
/// | `KAFKA_BROKERS` | bootstrap servers (comma-separated) |
/// | `KAFKA_SECURITY_PROTOCOL` | `PLAINTEXT` \| `SASL_SSL` \| `SSL` \| `SASL_PLAINTEXT` |
/// | `KAFKA_SASL_MECHANISM` | e.g. `PLAIN`, `SCRAM-SHA-512` |
/// | `KAFKA_SASL_USERNAME` / `KAFKA_SASL_PASSWORD` | SASL credentials |
/// | `KAFKA_SSL_CA_PEM` | inline CA certificate (PEM) for the trust store |
/// | `KAFKA_SESSION_TIMEOUT_MS` | consumer session timeout override (default 30000) |
#[derive(Clone, Debug)]
pub struct KafkaSettings {
    pub brokers: String,
    pub security_protocol: Option<String>,
    pub sasl_mechanism: Option<String>,
    pub sasl_username: Option<String>,
    pub sasl_password: Option<String>,
    pub ssl_ca_pem: Option<String>,
    pub session_timeout_ms: u32,
}

impl KafkaSettings {
    /// Read every setting from the environment. Empty strings are treated as unset.
    pub fn from_env() -> Self {
        let e = |k: &str| std::env::var(k).ok().filter(|s| !s.is_empty());
        Self {
            brokers: e("KAFKA_BROKERS").unwrap_or_default(),
            security_protocol: e("KAFKA_SECURITY_PROTOCOL"),
            sasl_mechanism: e("KAFKA_SASL_MECHANISM"),
            sasl_username: e("KAFKA_SASL_USERNAME"),
            sasl_password: e("KAFKA_SASL_PASSWORD"),
            ssl_ca_pem: e("KAFKA_SSL_CA_PEM"),
            session_timeout_ms: e("KAFKA_SESSION_TIMEOUT_MS")
                .and_then(|s| s.parse().ok())
                .unwrap_or(30_000),
        }
    }

    /// Bootstrap servers + SASL/SSL as key/value settings — the single platform
    /// replacement for the per-service `apply_kafka_security` copies. When
    /// `security_protocol` is unset the client stays PLAINTEXT (dev).
    pub fn security_settings(&self) -> Vec<Setting> {
        let mut s = vec![("bootstrap.servers", self.brokers.clone())];
        if let Some(v) = &self.security_protocol {
            s.push(("security.protocol", v.clone()));
        }
        if let Some(v) = &self.sasl_mechanism {
            s.push(("sasl.mechanism", v.clone()));
        }
        if let Some(v) = &self.sasl_username {
            s.push(("sasl.username", v.clone()));
        }
        if let Some(v) = &self.sasl_password {
            s.push(("sasl.password", v.clone()));
        }
        if let Some(v) = &self.ssl_ca_pem {
            // Inline PEM (mounted from a secret as an env var) rather than a file path.
            s.push(("ssl.ca.pem", v.clone()));
        }
        s
    }
}