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

//! Standardized at-least-once consumer settings.

use super::settings::{KafkaSettings, Setting};

/// Where a brand-new consumer group starts.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OffsetReset {
    /// Replay the topic's retained history (audit/state-rebuild consumers).
    Earliest,
    /// Only new messages from now on (snapshot/live consumers).
    Latest,
}

impl OffsetReset {
    pub fn as_str(self) -> &'static str {
        match self {
            OffsetReset::Earliest => "earliest",
            OffsetReset::Latest => "latest",
        }
    }
}

/// Consumer settings with the platform's standard **at-least-once** semantics
/// (apply to your own `rdkafka::ClientConfig`; also set `client.id = <service>`):
///
/// - `enable.auto.commit=false` + `enable.auto.offset.store=false` — commit only
///   after processing (the app owns commit).
/// - `isolation.level=read_committed` — never see aborted transactional writes.
/// - `auto.offset.reset` — explicit per consumer (see [`OffsetReset`]).
///
/// `group_id` is owned in code as `<service>[.<purpose>]`.
pub fn consumer_settings(
    settings: &KafkaSettings,
    group_id: &str,
    offset_reset: OffsetReset,
) -> Vec<Setting> {
    let mut s = settings.security_settings();
    s.extend([
        ("group.id", group_id.to_string()),
        ("enable.auto.commit", "false".to_string()),
        ("enable.auto.offset.store", "false".to_string()),
        ("enable.partition.eof", "false".to_string()),
        ("isolation.level", "read_committed".to_string()),
        ("auto.offset.reset", offset_reset.as_str().to_string()),
        ("session.timeout.ms", settings.session_timeout_ms.to_string()),
    ]);
    s
}