b453d5d
use serde::Deserialize;
const DEFAULT_NAME_SERVER: &str = "127.0.0.1:9876";
const DEFAULT_PRODUCER_GROUP: &str = "gitdata";
const DEFAULT_TOPIC_PREFIX: &str = "gitdata";
const DEFAULT_SEND_TIMEOUT_MS: u64 = 5_000;
const DEFAULT_MAX_MESSAGE_BYTES: usize = 4 * 1024 * 1024;
const MAX_TOPIC_LEN: usize = 127;
fn default_name_server() -> String {
DEFAULT_NAME_SERVER.to_string()
}
fn default_producer_group() -> String {
DEFAULT_PRODUCER_GROUP.to_string()
}
fn default_topic_prefix() -> String {
DEFAULT_TOPIC_PREFIX.to_string()
}
fn default_send_timeout_ms() -> u64 {
DEFAULT_SEND_TIMEOUT_MS
}
fn default_max_message_bytes() -> usize {
DEFAULT_MAX_MESSAGE_BYTES
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
#[serde(default)]
pub struct RocketMqConfig {
pub enabled: bool,
#[serde(default = "default_name_server")]
pub name_server: String,
#[serde(default = "default_producer_group")]
pub producer_group: String,
#[serde(default = "default_topic_prefix")]
pub topic_prefix: String,
#[serde(default = "default_send_timeout_ms")]
pub send_timeout_ms: u64,
#[serde(default = "default_max_message_bytes")]
pub max_message_bytes: usize,
}
impl Default for RocketMqConfig {
fn default() -> Self {
Self {
enabled: false,
name_server: default_name_server(),
producer_group: default_producer_group(),
topic_prefix: default_topic_prefix(),
send_timeout_ms: default_send_timeout_ms(),
max_message_bytes: default_max_message_bytes(),
}
}
}
impl RocketMqConfig {
pub fn name_servers(&self) -> Vec<&str> {
self.name_server
.split([';', ','])
.map(str::trim)
.filter(|server| !server.is_empty())
.collect()
}
pub fn topic(&self, area: &str) -> String {
format!("{}-{area}", self.topic_prefix)
}
pub fn is_valid(&self) -> Result<(), String> {
if !self.enabled {
return Ok(());
}
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.producer_group) {
return Err("rocketmq.producer_group must use letters, digits, `_` or `-`".to_string());
}
if !is_valid_identifier(&self.topic_prefix) {
return Err("rocketmq.topic_prefix must use letters, digits, `_` or `-`".to_string());
}
if self.topic_prefix.len() + 1 + 64 > MAX_TOPIC_LEN {
return Err(format!(
"rocketmq.topic_prefix leaves no room for a topic under {MAX_TOPIC_LEN} characters"
));
}
if self.send_timeout_ms == 0 {
return Err("rocketmq.send_timeout_ms must be greater than zero".to_string());
}
if self.max_message_bytes == 0 {
return Err("rocketmq.max_message_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, '_' | '-'))
}