use std::path::PathBuf;
use serde::{Deserialize, Serialize};
use crate::io::FileWriterConfig;
pub use crate::io::RotationPeriod;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DlqMode {
#[default]
Cascade,
FanOut,
FileOnly,
KafkaOnly,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct DlqConfig {
pub enabled: bool,
pub mode: DlqMode,
pub file: FileDlqConfig,
#[cfg(feature = "dlq-kafka")]
pub kafka: KafkaDlqConfig,
}
impl Default for DlqConfig {
fn default() -> Self {
Self {
enabled: true,
mode: DlqMode::default(),
file: FileDlqConfig::default(),
#[cfg(feature = "dlq-kafka")]
kafka: KafkaDlqConfig::default(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct FileDlqConfig {
pub enabled: bool,
pub path: PathBuf,
pub rotation: RotationPeriod,
pub max_age_days: u32,
pub compress_rotated: bool,
}
impl Default for FileDlqConfig {
fn default() -> Self {
Self {
enabled: true,
path: PathBuf::from("/var/spool/dfe/dlq"),
rotation: RotationPeriod::default(),
max_age_days: 30,
compress_rotated: true,
}
}
}
impl FileDlqConfig {
#[must_use]
pub fn to_writer_config(&self) -> FileWriterConfig {
FileWriterConfig {
path: self.path.clone(),
rotation: self.rotation,
max_age_days: self.max_age_days,
compress_rotated: self.compress_rotated,
}
}
}
#[cfg(feature = "dlq-kafka")]
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct KafkaDlqConfig {
pub enabled: bool,
pub routing: DlqRouting,
pub topic_suffix: String,
pub common_topic: String,
pub send_timeout_ms: u64,
}
#[cfg(feature = "dlq-kafka")]
impl Default for KafkaDlqConfig {
fn default() -> Self {
Self {
enabled: true,
routing: DlqRouting::default(),
topic_suffix: ".dlq".to_string(),
common_topic: "dfe.dlq".to_string(),
send_timeout_ms: 5000,
}
}
}
#[cfg(feature = "dlq-kafka")]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DlqRouting {
#[default]
PerTable,
Common,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_config_defaults() {
let config = DlqConfig::default();
assert!(config.enabled);
assert_eq!(config.mode, DlqMode::Cascade);
assert!(config.file.enabled);
assert_eq!(config.file.max_age_days, 30);
assert!(config.file.compress_rotated);
assert_eq!(config.file.rotation, RotationPeriod::Hourly);
}
#[test]
fn test_config_serde_roundtrip() {
let config = DlqConfig {
enabled: true,
mode: DlqMode::FanOut,
file: FileDlqConfig {
enabled: true,
path: "/tmp/test-dlq".into(),
rotation: RotationPeriod::Daily,
max_age_days: 7,
compress_rotated: false,
},
#[cfg(feature = "dlq-kafka")]
kafka: KafkaDlqConfig::default(),
};
let json = serde_json::to_string(&config).expect("serialise");
let parsed: DlqConfig = serde_json::from_str(&json).expect("deserialise");
assert_eq!(parsed.mode, DlqMode::FanOut);
assert_eq!(parsed.file.rotation, RotationPeriod::Daily);
assert_eq!(parsed.file.max_age_days, 7);
}
#[test]
fn test_dlq_mode_serde() {
let json = r#""cascade""#;
let mode: DlqMode = serde_json::from_str(json).expect("deserialise");
assert_eq!(mode, DlqMode::Cascade);
let json = r#""fan_out""#;
let mode: DlqMode = serde_json::from_str(json).expect("deserialise");
assert_eq!(mode, DlqMode::FanOut);
}
}