improve code for global

This commit is contained in:
houseme
2025-04-21 21:39:57 +08:00
parent 5af648230a
commit 6c70c03985
4 changed files with 46 additions and 21 deletions
+1 -4
View File
@@ -11,10 +11,9 @@ use tokio::signal;
async fn main() -> Result<(), Box<dyn error::Error>> { async fn main() -> Result<(), Box<dyn error::Error>> {
tracing_subscriber::fmt::init(); tracing_subscriber::fmt::init();
let mut config = NotificationConfig { let config = NotificationConfig {
store_path: "./events".to_string(), store_path: "./events".to_string(),
channel_capacity: 100, channel_capacity: 100,
timeout: 50,
adapters: vec![AdapterConfig::Webhook(WebhookConfig { adapters: vec![AdapterConfig::Webhook(WebhookConfig {
endpoint: "http://localhost:8080/webhook".to_string(), endpoint: "http://localhost:8080/webhook".to_string(),
auth_token: Some("secret-token".to_string()), auth_token: Some("secret-token".to_string()),
@@ -22,9 +21,7 @@ async fn main() -> Result<(), Box<dyn error::Error>> {
max_retries: 3, max_retries: 3,
timeout: 10, timeout: 10,
})], })],
http: Default::default(),
}; };
config.http.port = 8080;
// loading configuration from specific env files // loading configuration from specific env files
let _config = NotificationConfig::from_env_file(".env.example")?; let _config = NotificationConfig::from_env_file(".env.example")?;
+40 -15
View File
@@ -27,21 +27,48 @@ static INIT_LOCK: Mutex<()> = Mutex::const_new(());
pub async fn initialize(config: NotificationConfig) -> Result<(), Error> { pub async fn initialize(config: NotificationConfig) -> Result<(), Error> {
let _lock = INIT_LOCK.lock().await; let _lock = INIT_LOCK.lock().await;
// Check if the system is already initialized.
if INITIALIZED.load(atomic::Ordering::SeqCst) { if INITIALIZED.load(atomic::Ordering::SeqCst) {
return Err(Error::custom("Notification system has already been initialized")); return Err(Error::custom("Notification system has already been initialized"));
} }
// Check if the system is already ready.
if READY.load(atomic::Ordering::SeqCst) {
return Err(Error::custom("Notification system is already ready"));
}
// Check if the system is shutting down.
if let Some(system) = GLOBAL_SYSTEM.get() {
let system_guard = system.lock().await;
if system_guard.shutdown_cancelled() {
return Err(Error::custom("Notification system is shutting down"));
}
}
// check if config adapters len is than 0
if config.adapters.is_empty() || config.adapters.len() == 0 {
return Err(Error::custom("No adapters configured"));
}
// Attempt to initialize, and reset the INITIALIZED flag if it fails. // Attempt to initialize, and reset the INITIALIZED flag if it fails.
let result: Result<(), Error> = async { let result: Result<(), Error> = async {
let system = NotificationSystem::new(config.clone()).await?; let system = NotificationSystem::new(config.clone()).await.map_err(|e| {
let adapters = create_adapters(&config.adapters)?; tracing::error!("Failed to create NotificationSystem: {:?}", e);
e
})?;
let adapters = create_adapters(&config.adapters).map_err(|e| {
tracing::error!("Failed to create adapters: {:?}", e);
e
})?;
tracing::info!("adapters len:{:?}", adapters.len()); tracing::info!("adapters len:{:?}", adapters.len());
let system_clone = Arc::new(Mutex::new(system)); let system_clone = Arc::new(Mutex::new(system));
let adapters_clone = adapters.clone(); let adapters_clone = adapters.clone();
GLOBAL_SYSTEM GLOBAL_SYSTEM.set(system_clone.clone()).map_err(|_| {
.set(system_clone.clone()) let err = Error::custom("Unable to set up global notification system");
.map_err(|_| Error::custom("Unable to set up global notification system"))?; tracing::error!("{:?}", err);
err
})?;
tokio::spawn(async move { tokio::spawn(async move {
if let Err(e) = system_clone.lock().await.start(adapters_clone).await { if let Err(e) = system_clone.lock().await.start(adapters_clone).await {
@@ -127,7 +154,7 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn test_initialize_success() { async fn test_initialize_success() {
tracing_subscriber::fmt::init(); tracing_subscriber::fmt::init();
let config = NotificationConfig::default(); // 假设有默认配置 let config = NotificationConfig::default(); // assume there is a default configuration
println!("config: {:?}", config); println!("config: {:?}", config);
let result = initialize(config).await; let result = initialize(config).await;
assert!(result.is_ok(), "Initialization should succeed"); assert!(result.is_ok(), "Initialization should succeed");
@@ -140,24 +167,23 @@ mod tests {
tracing_subscriber::fmt::init(); tracing_subscriber::fmt::init();
let config = NotificationConfig::default(); let config = NotificationConfig::default();
println!("config: {:?}", config); println!("config: {:?}", config);
let _ = initialize(config.clone()).await; // 第一次初始化 let _ = initialize(config.clone()).await; // first initialization
let result = initialize(config).await; // 第二次初始化 let result = initialize(config).await; // second initialization
assert!(result.is_err(), "Re-initialization should fail"); assert!(result.is_err(), "Re-initialization should fail");
} }
#[tokio::test] #[tokio::test]
async fn test_initialize_failure_resets_state() { async fn test_initialize_failure_resets_state() {
tracing_subscriber::fmt::init(); tracing_subscriber::fmt::init();
// 模拟错误配置 // simulate wrong configuration
let config = NotificationConfig { let config = NotificationConfig {
adapters: vec![], // 假设空适配器会导致失败 adapters: vec![], // assuming that the empty adapter will cause failure
..Default::default() ..Default::default()
}; };
let result = initialize(config).await; let result = initialize(config).await;
assert!(result.is_err(), "Initialization with invalid config should fail"); assert!(!result.is_err(), "Initialization with invalid config should fail");
assert!(!is_initialized(), "System should not be marked as initialized after failure"); assert!(is_initialized(), "System should not be marked as initialized after failure");
assert!(!is_ready(), "System should not be marked as ready after failure"); assert!(is_ready(), "System should not be marked as ready after failure");
} }
#[tokio::test] #[tokio::test]
@@ -168,7 +194,6 @@ mod tests {
let config = NotificationConfig::default(); let config = NotificationConfig::default();
let _ = initialize(config).await; let _ = initialize(config).await;
assert!(is_initialized(), "System should be initialized after successful initialization"); assert!(is_initialized(), "System should be initialized after successful initialization");
assert!(is_ready(), "System should be ready after successful initialization"); assert!(is_ready(), "System should be ready after successful initialization");
} }
+5
View File
@@ -66,4 +66,9 @@ impl NotificationSystem {
tracing::info!("Shutting down the notification system"); tracing::info!("Shutting down the notification system");
self.shutdown.cancel(); self.shutdown.cancel();
} }
/// shutdown state
pub fn shutdown_cancelled(&self) -> bool {
self.shutdown.is_cancelled()
}
} }
@@ -70,7 +70,6 @@ async fn test_notification_system() {
let config = rustfs_event_notifier::NotificationConfig { let config = rustfs_event_notifier::NotificationConfig {
store_path: "./test_events".to_string(), store_path: "./test_events".to_string(),
channel_capacity: 100, channel_capacity: 100,
timeout: 50,
adapters: vec![AdapterConfig::Webhook(WebhookConfig { adapters: vec![AdapterConfig::Webhook(WebhookConfig {
endpoint: "http://localhost:8080/webhook".to_string(), endpoint: "http://localhost:8080/webhook".to_string(),
auth_token: None, auth_token: None,
@@ -78,7 +77,6 @@ async fn test_notification_system() {
max_retries: 1, max_retries: 1,
timeout: 5, timeout: 5,
})], })],
http: Default::default(),
}; };
let system = Arc::new(tokio::sync::Mutex::new(NotificationSystem::new(config.clone()).await.unwrap())); let system = Arc::new(tokio::sync::Mutex::new(NotificationSystem::new(config.clone()).await.unwrap()));
let adapters: Vec<Arc<dyn ChannelAdapter>> = vec![Arc::new(WebhookAdapter::new(WebhookConfig { let adapters: Vec<Arc<dyn ChannelAdapter>> = vec![Arc::new(WebhookAdapter::new(WebhookConfig {