road-runner-common 0.8.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
82
83
84
85
86

//! Standardized Kafka configuration for platform services (feature = `kafka`).
//!
//! Returns plain `(key, value)` settings — never a version-specific
//! `rdkafka::ClientConfig` — because services span rdkafka 0.36–0.39. Each service
//! applies them to its own config:
//!
//! ```ignore
//! use road_runner_common::kafka::{KafkaSettings, consumer_settings, OffsetReset};
//!
//! let settings = KafkaSettings::from_env();
//! let mut cfg = rdkafka::ClientConfig::new();
//! for (k, v) in consumer_settings(&settings, "cex-history", OffsetReset::Earliest) {
//!     cfg.set(k, v);
//! }
//! cfg.set("client.id", "cex-history");
//! let consumer: StreamConsumer = cfg.create()?;
//! ```
//!
//! Config layering: **ENV** → connection+security ([`KafkaSettings::from_env`]);
//! **CODE** → topics/`group.id`/`client.id`/semantics; **IaC** → topic lifecycle.

pub mod consumer;
pub mod metrics;
pub mod producer;
pub mod settings;

pub use consumer::{consumer_settings, OffsetReset};
pub use producer::producer_settings;
pub use settings::{KafkaSettings, Setting};

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    fn settings() -> KafkaSettings {
        KafkaSettings {
            brokers: "b:9092".into(),
            security_protocol: Some("SASL_SSL".into()),
            sasl_mechanism: Some("PLAIN".into()),
            sasl_username: Some("u".into()),
            sasl_password: Some("p".into()),
            ssl_ca_pem: None,
            session_timeout_ms: 30_000,
        }
    }

    fn map(s: Vec<Setting>) -> HashMap<&'static str, String> {
        s.into_iter().collect()
    }

    #[test]
    fn producer_is_durable_and_idempotent() {
        let m = map(producer_settings(&settings()));
        assert_eq!(m["acks"], "all");
        assert_eq!(m["enable.idempotence"], "true");
        assert_eq!(m["security.protocol"], "SASL_SSL");
        assert_eq!(m["bootstrap.servers"], "b:9092");
    }

    #[test]
    fn consumer_is_manual_commit_read_committed() {
        let m = map(consumer_settings(&settings(), "cex-history", OffsetReset::Earliest));
        assert_eq!(m["group.id"], "cex-history");
        assert_eq!(m["enable.auto.commit"], "false");
        assert_eq!(m["isolation.level"], "read_committed");
        assert_eq!(m["auto.offset.reset"], "earliest");
        assert_eq!(m["sasl.username"], "u");
    }

    #[test]
    fn dev_plaintext_needs_only_brokers() {
        let s = KafkaSettings {
            brokers: "b:9092".into(),
            security_protocol: None,
            sasl_mechanism: None,
            sasl_username: None,
            sasl_password: None,
            ssl_ca_pem: None,
            session_timeout_ms: 30_000,
        };
        let m = map(consumer_settings(&s, "g", OffsetReset::Latest));
        assert!(!m.contains_key("security.protocol"));
        assert_eq!(m["auto.offset.reset"], "latest");
    }
}