main
use serde::{Deserialize, Serialize};
pub const DEFAULT_NAME_SERVER: &str = "127.0.0.1:9876";
pub const DEFAULT_CONSUMER_GROUP: &str = "kenevent";
pub const DEFAULT_TOPIC_PREFIXES: [&str; 2] = ["gitdata", "kenshell"];
pub const DEFAULT_CONSUME_TIMEOUT_MS: u64 = 15 * 60 * 1_000;
pub const DEFAULT_MAX_RECONSUME_TIMES: u32 = 16;
pub const DEFAULT_PULL_BATCH_SIZE: u32 = 32;
pub const DEFAULT_PULL_BATCH_SIZE_IN_BYTES: u32 = 256 * 1024;
pub const DEFAULT_PULL_INTERVAL_MS: u64 = 0;
pub const MAX_TOPIC_LEN: usize = 127;
pub const MAX_AREA_LEN: usize = 64;
fn default_name_server() -> String {
DEFAULT_NAME_SERVER.to_string()
}
fn default_consumer_group() -> String {
DEFAULT_CONSUMER_GROUP.to_string()
}
fn default_topic_prefixes() -> Vec<String> {
DEFAULT_TOPIC_PREFIXES
.iter()
.map(|prefix| (*prefix).to_string())
.collect()
}
fn default_consume_timeout_ms() -> u64 {
DEFAULT_CONSUME_TIMEOUT_MS
}
fn default_max_reconsume_times() -> u32 {
DEFAULT_MAX_RECONSUME_TIMES
}
fn default_pull_batch_size() -> u32 {
DEFAULT_PULL_BATCH_SIZE
}
fn default_pull_batch_size_in_bytes() -> u32 {
DEFAULT_PULL_BATCH_SIZE_IN_BYTES
}
fn default_pull_interval_ms() -> u64 {
DEFAULT_PULL_INTERVAL_MS
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct RocketMqConsumerConfig {
#[serde(default = "default_name_server")]
pub name_server: String,
#[serde(default = "default_consumer_group")]
pub consumer_group: String,
#[serde(default = "default_topic_prefixes")]
pub topic_prefixes: Vec<String>,
#[serde(default = "default_consume_timeout_ms")]
pub consume_timeout_ms: u64,
#[serde(default = "default_max_reconsume_times")]
pub max_reconsume_times: u32,
#[serde(default = "default_pull_batch_size")]
pub pull_batch_size: u32,
#[serde(default = "default_pull_batch_size_in_bytes")]
pub pull_batch_size_in_bytes: u32,
#[serde(default = "default_pull_interval_ms")]
pub pull_interval_ms: u64,
}
impl Default for RocketMqConsumerConfig {
fn default() -> Self {
Self {
name_server: default_name_server(),
consumer_group: default_consumer_group(),
topic_prefixes: default_topic_prefixes(),
consume_timeout_ms: default_consume_timeout_ms(),
max_reconsume_times: default_max_reconsume_times(),
pull_batch_size: default_pull_batch_size(),
pull_batch_size_in_bytes: default_pull_batch_size_in_bytes(),
pull_interval_ms: default_pull_interval_ms(),
}
}
}
impl RocketMqConsumerConfig {
pub fn name_servers(&self) -> Vec<&str> {
self.name_server
.split([';', ','])
.map(str::trim)
.filter(|server| !server.is_empty())
.collect()
}
pub fn consume_timeout_minutes(&self) -> u64 {
self.consume_timeout_ms.div_ceil(60_000).max(1)
}
pub fn validate(&self) -> Result<(), String> {
let servers = self.name_servers();
if servers.is_empty() {
return Err("rocketmq.name_server must not be blank".to_string());
}
for server in servers {
let (host, port) = server.rsplit_once(':').ok_or_else(|| {
format!("rocketmq.name_server `{server}` must use the `host:port` form")
})?;
if host.trim().is_empty() {
return Err(format!("rocketmq.name_server `{server}` has an empty host"));
}
let port = port
.parse::<u16>()
.map_err(|_| format!("rocketmq.name_server `{server}` has an invalid port"))?;
if port == 0 {
return Err(format!("rocketmq.name_server `{server}` has a zero port"));
}
}
if !is_valid_identifier(&self.consumer_group) {
return Err("rocketmq.consumer_group must use letters, digits, `_` or `-`".to_string());
}
if self.topic_prefixes.is_empty() {
return Err("rocketmq.topic_prefixes must not be empty".to_string());
}
for (index, prefix) in self.topic_prefixes.iter().enumerate() {
if !DEFAULT_TOPIC_PREFIXES.contains(&prefix.as_str()) {
return Err(format!(
"unsupported rocketmq.topic_prefixes[{index}]: {prefix}"
));
}
if !is_valid_identifier(prefix) {
return Err(format!(
"rocketmq.topic_prefixes[{index}] must use letters, digits, `_` or `-`"
));
}
if prefix.len() + 1 + MAX_AREA_LEN > MAX_TOPIC_LEN {
return Err(format!(
"rocketmq.topic_prefixes[{index}] leaves no room for a topic under {MAX_TOPIC_LEN} characters"
));
}
if self.topic_prefixes[..index].contains(prefix) {
return Err(format!(
"rocketmq.topic_prefixes[{index}] `{prefix}` is duplicated"
));
}
}
if self.consume_timeout_ms == 0 {
return Err("rocketmq.consume_timeout_ms must be greater than zero".to_string());
}
if self.max_reconsume_times == 0 {
return Err("rocketmq.max_reconsume_times must be greater than zero".to_string());
}
if self.max_reconsume_times > i32::MAX as u32 {
return Err("rocketmq.max_reconsume_times exceeds i32::MAX".to_string());
}
if self.pull_batch_size == 0 {
return Err("rocketmq.pull_batch_size must be greater than zero".to_string());
}
if self.pull_batch_size_in_bytes == 0 {
return Err("rocketmq.pull_batch_size_in_bytes must be greater than zero".to_string());
}
Ok(())
}
}
fn is_valid_identifier(value: &str) -> bool {
!value.is_empty()
&& value.len() <= MAX_TOPIC_LEN
&& value
.chars()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-'))
}