From 5af648230af0999f4713612b2e900c73b269b92f Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 21 Apr 2025 19:01:31 +0800 Subject: [PATCH] improve code --- Cargo.lock | 1 - crates/event-notifier/Cargo.toml | 2 - crates/event-notifier/src/adapter/mod.rs | 10 +- crates/event-notifier/src/config.rs | 30 --- crates/event-notifier/src/error.rs | 2 +- crates/event-notifier/src/global.rs | 297 ++++++++++------------- crates/event-notifier/src/lib.rs | 8 +- crates/event-notifier/src/notifier.rs | 38 +-- crates/event-notifier/src/producer.rs | 83 ------- 9 files changed, 139 insertions(+), 332 deletions(-) delete mode 100644 crates/event-notifier/src/producer.rs diff --git a/Cargo.lock b/Cargo.lock index 8aabcc753..917fd1444 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7350,7 +7350,6 @@ name = "rustfs-event-notifier" version = "0.0.1" dependencies = [ "async-trait", - "axum", "dotenvy", "figment", "rdkafka", diff --git a/crates/event-notifier/Cargo.toml b/crates/event-notifier/Cargo.toml index 0c3bc3947..ce5b1f04b 100644 --- a/crates/event-notifier/Cargo.toml +++ b/crates/event-notifier/Cargo.toml @@ -11,11 +11,9 @@ default = ["webhook"] webhook = ["dep:reqwest"] kafka = ["rdkafka"] mqtt = ["rumqttc"] -http-producer = ["dep:axum"] [dependencies] async-trait = { workspace = true } -axum = { workspace = true, optional = true } dotenvy = { workspace = true } figment = { workspace = true, features = ["toml", "yaml", "env"] } rdkafka = { workspace = true, features = ["tokio"], optional = true } diff --git a/crates/event-notifier/src/adapter/mod.rs b/crates/event-notifier/src/adapter/mod.rs index 9b8cab9f0..fa12aa97b 100644 --- a/crates/event-notifier/src/adapter/mod.rs +++ b/crates/event-notifier/src/adapter/mod.rs @@ -21,14 +21,14 @@ pub trait ChannelAdapter: Send + Sync + 'static { } /// Creates channel adapters based on the provided configuration. -pub fn create_adapters(configs: &[AdapterConfig]) -> Result>, Box> { +pub fn create_adapters(configs: &[AdapterConfig]) -> Result>, Error> { let mut adapters: Vec> = Vec::new(); for config in configs { match config { #[cfg(feature = "webhook")] AdapterConfig::Webhook(webhook_config) => { - webhook_config.validate().map_err(|e| Box::new(Error::ConfigError(e)))?; + webhook_config.validate().map_err(Error::ConfigError)?; adapters.push(Arc::new(webhook::WebhookAdapter::new(webhook_config.clone()))); } #[cfg(feature = "kafka")] @@ -42,11 +42,11 @@ pub fn create_adapters(configs: &[AdapterConfig]) -> Result return Err(Box::new(Error::FeatureDisabled("webhook"))), + AdapterConfig::Webhook(_) => return Err(Error::FeatureDisabled("webhook")), #[cfg(not(feature = "kafka"))] - AdapterConfig::Kafka(_) => return Err(Box::new(Error::FeatureDisabled("kafka"))), + AdapterConfig::Kafka(_) => return Err(Error::FeatureDisabled("kafka")), #[cfg(not(feature = "mqtt"))] - AdapterConfig::Mqtt(_) => return Err(Box::new(Error::FeatureDisabled("mqtt"))), + AdapterConfig::Mqtt(_) => return Err(Error::FeatureDisabled("mqtt")), } } diff --git a/crates/event-notifier/src/config.rs b/crates/event-notifier/src/config.rs index af5f88998..c648ceac3 100644 --- a/crates/event-notifier/src/config.rs +++ b/crates/event-notifier/src/config.rs @@ -63,25 +63,6 @@ pub enum AdapterConfig { Mqtt(MqttConfig), } -/// http producer configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct HttpProducerConfig { - #[serde(default = "default_http_port")] - pub port: u16, -} - -impl Default for HttpProducerConfig { - fn default() -> Self { - Self { - port: default_http_port(), - } - } -} - -fn default_http_port() -> u16 { - 3000 -} - /// Configuration for the notification system. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NotificationConfig { @@ -89,11 +70,7 @@ pub struct NotificationConfig { pub store_path: String, #[serde(default = "default_channel_capacity")] pub channel_capacity: usize, - #[serde(default = "default_timeout")] - pub timeout: u64, pub adapters: Vec, - #[serde(default)] - pub http: HttpProducerConfig, } impl Default for NotificationConfig { @@ -101,9 +78,7 @@ impl Default for NotificationConfig { Self { store_path: default_store_path(), channel_capacity: default_channel_capacity(), - timeout: default_timeout(), adapters: Vec::new(), - http: HttpProducerConfig::default(), } } } @@ -157,8 +132,3 @@ fn default_store_path() -> String { fn default_channel_capacity() -> usize { 10000 // Reasonable default values for high concurrency systems } - -/// Provides the recommended default timeout for high concurrency systems -fn default_timeout() -> u64 { - 50 // Reasonable default values for high concurrency systems -} diff --git a/crates/event-notifier/src/error.rs b/crates/event-notifier/src/error.rs index 9f1e41b4b..7a7d3192e 100644 --- a/crates/event-notifier/src/error.rs +++ b/crates/event-notifier/src/error.rs @@ -30,7 +30,7 @@ pub enum Error { MissingField(&'static str), #[error("field verification failed:{0}")] ValidationError(&'static str), - #[error("{0}")] + #[error("Custom error: {0}")] Custom(String), #[error("Configuration error: {0}")] ConfigError(String), diff --git a/crates/event-notifier/src/global.rs b/crates/event-notifier/src/global.rs index ed666a8d2..3476ce45c 100644 --- a/crates/event-notifier/src/global.rs +++ b/crates/event-notifier/src/global.rs @@ -1,7 +1,5 @@ -use crate::{ChannelAdapter, Error, Event, NotificationConfig, NotificationSystem}; +use crate::{create_adapters, Error, Event, NotificationConfig, NotificationSystem}; use std::sync::{atomic, Arc}; -use std::time; -use std::time::Duration; use tokio::sync::{Mutex, OnceCell}; static GLOBAL_SYSTEM: OnceCell>> = OnceCell::const_new(); @@ -9,206 +7,169 @@ static INITIALIZED: atomic::AtomicBool = atomic::AtomicBool::new(false); static READY: atomic::AtomicBool = atomic::AtomicBool::new(false); static INIT_LOCK: Mutex<()> = Mutex::const_new(()); -/// initialize the global notification system +/// Initializes the global notification system. +/// +/// This function performs the following steps: +/// 1. Checks if the system is already initialized. +/// 2. Creates a new `NotificationSystem` instance. +/// 3. Creates adapters based on the provided configuration. +/// 4. Starts the notification system with the created adapters. +/// 5. Sets the global system instance. +/// +/// # Errors +/// +/// Returns an error if: +/// - The system is already initialized. +/// - Creating the `NotificationSystem` fails. +/// - Creating adapters fails. +/// - Starting the notification system fails. +/// - Setting the global system instance fails. pub async fn initialize(config: NotificationConfig) -> Result<(), Error> { - // use lock to protect the initialization process - let _lock = INIT_LOCK.lock().await; - - if INITIALIZED.swap(true, atomic::Ordering::SeqCst) { - return Err(Error::custom("notify the system has been initialized")); - } - - match Arc::new(Mutex::new(NotificationSystem::new(config).await?)) { - system => { - if let Err(_) = GLOBAL_SYSTEM.set(system.clone()) { - INITIALIZED.store(false, atomic::Ordering::SeqCst); - return Err(Error::custom("unable to set up global notification system")); - } - system - } - }; - Ok(()) -} - -/// securely initialize the global notification system -pub async fn initialize_safe(config: NotificationConfig) -> Result<(), Error> { - // use-lock-to-protect-the-initialization-process let _lock = INIT_LOCK.lock().await; if INITIALIZED.load(atomic::Ordering::SeqCst) { - return Err(Error::custom("notify the system has been initialized")); + return Err(Error::custom("Notification system has already been initialized")); + } + + // Attempt to initialize, and reset the INITIALIZED flag if it fails. + let result: Result<(), Error> = async { + let system = NotificationSystem::new(config.clone()).await?; + let adapters = create_adapters(&config.adapters)?; + tracing::info!("adapters len:{:?}", adapters.len()); + let system_clone = Arc::new(Mutex::new(system)); + let adapters_clone = adapters.clone(); + + GLOBAL_SYSTEM + .set(system_clone.clone()) + .map_err(|_| Error::custom("Unable to set up global notification system"))?; + + tokio::spawn(async move { + if let Err(e) = system_clone.lock().await.start(adapters_clone).await { + tracing::error!("Notification system failed to start: {}", e); + } + tracing::info!("Notification system started in background"); + }); + tracing::info!("system start success,start set READY value"); + + READY.store(true, atomic::Ordering::SeqCst); + tracing::info!("Notification system is ready to process events"); + + Ok(()) + } + .await; + + if result.is_err() { + INITIALIZED.store(false, atomic::Ordering::SeqCst); + READY.store(false, atomic::Ordering::SeqCst); + return result; } - // set-initialization-flag INITIALIZED.store(true, atomic::Ordering::SeqCst); - - match Arc::new(Mutex::new(NotificationSystem::new(config).await?)) { - system => { - if let Err(_) = GLOBAL_SYSTEM.set(system.clone()) { - INITIALIZED.store(false, atomic::Ordering::SeqCst); - return Err(Error::custom("unable to set up global notification system")); - } - system - } - }; Ok(()) } -/// securely start the global notification system -pub async fn start_safe(adapters: Vec>) -> Result<(), Error> { - // start process with lock protection - let _lock = INIT_LOCK.lock().await; - - if !INITIALIZED.load(atomic::Ordering::SeqCst) { - return Err(Error::custom("notification system not initialized")); - } - - if READY.load(atomic::Ordering::SeqCst) { - return Err(Error::custom("notification system already started")); - } - - let system = get_system().await?; - - // Execute startup operations directly on the current thread, rather than generating new tasks - let mut system_guard = system.lock().await; - match system_guard.start(adapters).await { - Ok(_) => { - READY.store(true, atomic::Ordering::SeqCst); - tracing::info!("Notification system is ready to process events"); - Ok(()) - } - Err(e) => { - tracing::error!("Notify system start failed: {}", e); - INITIALIZED.store(false, atomic::Ordering::SeqCst); - Err(e) - } - } +/// Checks if the notification system is initialized. +pub fn is_initialized() -> bool { + INITIALIZED.load(atomic::Ordering::SeqCst) } -/// start the global notification system -pub async fn start(adapters: Vec>) -> Result<(), Error> { - let system = get_system().await?; - - // create a new task to run the system - let system_clone = Arc::clone(&system); - tokio::spawn(async move { - let mut system_guard = system_clone.lock().await; - match system_guard.start(adapters).await { - Ok(_) => { - // The system is started and runs normally, set the ready flag - READY.store(true, atomic::Ordering::SeqCst); - tracing::info!("Notification system is ready to process events"); - } - Err(e) => { - tracing::error!("Notify system start failed: {}", e); - INITIALIZED.store(false, atomic::Ordering::SeqCst); - } - } - }); - - // Wait for a while to ensure the system has a chance to start - tokio::time::sleep(Duration::from_millis(100)).await; - - Ok(()) +/// Checks if the notification system is ready. +pub fn is_ready() -> bool { + READY.load(atomic::Ordering::SeqCst) } -/// waiting for notification system to be fully ready -async fn wait_until_ready(timeout: Duration) -> Result<(), Error> { - let start = time::Instant::now(); - - while !READY.load(atomic::Ordering::SeqCst) { - if start.elapsed() > timeout { - return Err(Error::custom("timeout waiting for notification system to become ready")); - } - tokio::time::sleep(Duration::from_millis(50)).await; - } - - Ok(()) -} - -/// Initialize and start the global notification system +/// Sends an event to the notification system. /// -/// This method combines the functions of `initialize` and `start` to provide one-step setup: -/// - initialize system configuration -/// - create an adapter -/// - start event listening +/// # Errors /// -/// # Example -/// -/// ```rust -/// use rustfs_event_notifier::{initialize_and_start, NotificationConfig}; -/// -/// #[tokio::main] -/// async fn main() -> Result<(), rustfs_event_notifier::Error> { -/// let config = NotificationConfig { -/// store_path: "./events".to_string(), -/// channel_capacity: 100, -/// timeout: 0, -/// adapters: vec![/* 适配器配置 */], -/// http: Default::default(), -/// }; -/// -/// // complete initialization and startup in one step -/// initialize_and_start(config).await?; -/// Ok(()) -/// } -/// ``` -pub async fn initialize_and_start(config: NotificationConfig) -> Result<(), Error> { - // initialize the system first - initialize(config.clone()).await?; - - // create an adapter - let adapters = crate::create_adapters(&config.adapters).expect("failed to create adapters"); - - // start the system - start(adapters).await?; - - Ok(()) -} - -/// Initialize and start the global notification system and wait until it's ready -pub async fn initialize_and_start_with_ready_check(config: NotificationConfig, timeout: Duration) -> Result<(), Error> { - // initialize the system - initialize(config.clone()).await?; - - // create an adapter - let adapters = crate::create_adapters(&config.adapters).expect("failed to create adapters"); - - // start the system - start(adapters).await?; - - // wait for the system to be ready - wait_until_ready(timeout).await?; - - Ok(()) -} - -/// send events to notification system +/// Returns an error if: +/// - The system is not initialized. +/// - The system is not ready. +/// - Sending the event fails. pub async fn send_event(event: Event) -> Result<(), Error> { - // check if the system is ready to receive events if !READY.load(atomic::Ordering::SeqCst) { - return Err(Error::custom("notification system not ready, please wait for initialization to complete")); + return Err(Error::custom("Notification system not ready, please wait for initialization to complete")); } + let system = get_system().await?; let system_guard = system.lock().await; system_guard.send_event(event).await } -/// turn off the notification system +/// Shuts down the notification system. pub fn shutdown() -> Result<(), Error> { if let Some(system) = GLOBAL_SYSTEM.get() { let system_guard = system.blocking_lock(); system_guard.shutdown(); Ok(()) } else { - Err(Error::custom("notification system not initialized")) + Err(Error::custom("Notification system not initialized")) } } -/// get system instance +/// Retrieves the global notification system instance. +/// +/// # Errors +/// +/// Returns an error if the system is not initialized. async fn get_system() -> Result>, Error> { GLOBAL_SYSTEM .get() .cloned() - .ok_or_else(|| Error::custom("notification system not initialized")) + .ok_or_else(|| Error::custom("Notification system not initialized")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::NotificationConfig; + + #[tokio::test] + async fn test_initialize_success() { + tracing_subscriber::fmt::init(); + let config = NotificationConfig::default(); // 假设有默认配置 + println!("config: {:?}", config); + let result = initialize(config).await; + assert!(result.is_ok(), "Initialization should succeed"); + assert!(is_initialized(), "System should be marked as initialized"); + assert!(is_ready(), "System should be marked as ready"); + } + + #[tokio::test] + async fn test_initialize_twice() { + tracing_subscriber::fmt::init(); + let config = NotificationConfig::default(); + println!("config: {:?}", config); + let _ = initialize(config.clone()).await; // 第一次初始化 + let result = initialize(config).await; // 第二次初始化 + assert!(result.is_err(), "Re-initialization should fail"); + } + + #[tokio::test] + async fn test_initialize_failure_resets_state() { + tracing_subscriber::fmt::init(); + // 模拟错误配置 + let config = NotificationConfig { + adapters: vec![], // 假设空适配器会导致失败 + ..Default::default() + }; + + let result = initialize(config).await; + assert!(result.is_err(), "Initialization with invalid config should fail"); + assert!(!is_initialized(), "System should not be marked as initialized after failure"); + assert!(!is_ready(), "System should not be marked as ready after failure"); + } + + #[tokio::test] + async fn test_is_initialized_and_is_ready() { + tracing_subscriber::fmt::init(); + assert!(!is_initialized(), "System should not be initialized initially"); + assert!(!is_ready(), "System should not be ready initially"); + + let config = NotificationConfig::default(); + let _ = initialize(config).await; + + assert!(is_initialized(), "System should be initialized after successful initialization"); + assert!(is_ready(), "System should be ready after successful initialization"); + } } diff --git a/crates/event-notifier/src/lib.rs b/crates/event-notifier/src/lib.rs index 2fa9d8954..76981822c 100644 --- a/crates/event-notifier/src/lib.rs +++ b/crates/event-notifier/src/lib.rs @@ -5,7 +5,6 @@ mod error; mod event; mod global; mod notifier; -mod producer; mod store; pub use adapter::create_adapters; @@ -17,8 +16,6 @@ pub use adapter::mqtt::MqttAdapter; pub use adapter::webhook::WebhookAdapter; pub use adapter::ChannelAdapter; pub use bus::event_bus; -#[cfg(feature = "http-producer")] -pub use config::HttpProducerConfig; #[cfg(feature = "kafka")] pub use config::KafkaConfig; #[cfg(feature = "mqtt")] @@ -29,9 +26,6 @@ pub use config::{AdapterConfig, NotificationConfig}; pub use error::Error; pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source}; -pub use global::{ - initialize, initialize_and_start, initialize_and_start_with_ready_check, initialize_safe, send_event, shutdown, start, - start_safe, -}; +pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; pub use notifier::NotificationSystem; pub use store::EventStore; diff --git a/crates/event-notifier/src/notifier.rs b/crates/event-notifier/src/notifier.rs index e75fe25db..df389d878 100644 --- a/crates/event-notifier/src/notifier.rs +++ b/crates/event-notifier/src/notifier.rs @@ -1,10 +1,3 @@ -#[cfg(feature = "http-producer")] -pub use crate::producer::http::HttpProducer; -#[cfg(feature = "http-producer")] -pub use crate::producer::EventProducer; - -#[cfg(feature = "http-producer")] -pub use crate::config::HttpProducerConfig; use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotificationConfig}; use std::sync::Arc; use tokio::sync::mpsc; @@ -19,8 +12,6 @@ pub struct NotificationSystem { rx: Option>, store: Arc, shutdown: CancellationToken, - #[cfg(feature = "http-producer")] - http_config: HttpProducerConfig, } impl NotificationSystem { @@ -43,8 +34,6 @@ impl NotificationSystem { rx: Some(rx), store, shutdown, - #[cfg(feature = "http-producer")] - http_config: config.http, }) } @@ -55,28 +44,13 @@ impl NotificationSystem { let shutdown_clone = self.shutdown.clone(); let store_clone = self.store.clone(); - let bus_handle = tokio::spawn(async move { + tokio::spawn(async move { if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone).await { tracing::error!("Event bus failed: {}", e); } }); - #[cfg(feature = "http-producer")] - { - let producer = crate::producer::http::HttpProducer::new(self.tx.clone(), self.http_config.port); - producer.start().await?; - } - - tokio::select! { - result = bus_handle => { - result.map_err(Error::JoinError)?; - Ok(()) - }, - _ = self.shutdown.cancelled() => { - tracing::info!("System shutdown triggered"); - Ok(()) - } - } + Ok(()) } /// Sends an event to the notification system. @@ -89,13 +63,7 @@ impl NotificationSystem { /// Shuts down the notification system. /// This method is used to cancel the event bus and producer tasks. pub fn shutdown(&self) { + tracing::info!("Shutting down the notification system"); self.shutdown.cancel(); } - - /// Sets the HTTP port for the notification system. - /// This method is used to change the port for the HTTP producer. - #[cfg(feature = "http-producer")] - pub fn set_http_port(&mut self, port: u16) { - self.http_config.port = port; - } } diff --git a/crates/event-notifier/src/producer.rs b/crates/event-notifier/src/producer.rs deleted file mode 100644 index fc53b9462..000000000 --- a/crates/event-notifier/src/producer.rs +++ /dev/null @@ -1,83 +0,0 @@ -use crate::Error; -use crate::Event; -use async_trait::async_trait; - -/// event producer characteristics -#[allow(dead_code)] -#[async_trait] -pub trait EventProducer: Send + Sync { - /// start producer services - async fn start(&self) -> Result<(), Error>; - /// stop producer services - async fn stop(&self) -> Result<(), Error>; - /// send a single event - async fn send_event(&self, event: Event) -> Result<(), Error>; -} - -#[cfg(feature = "http-producer")] -pub mod http { - use super::*; - use axum::{routing::post, Json, Router}; - use std::sync::Arc; - use tokio::sync::mpsc; - - #[derive(Clone)] - pub struct HttpProducer { - tx: mpsc::Sender, - port: u16, - shutdown: Arc, - } - - impl HttpProducer { - pub fn new(tx: mpsc::Sender, port: u16) -> Self { - Self { - tx, - port, - shutdown: Arc::new(tokio::sync::Notify::new()), - } - } - } - - #[async_trait] - impl EventProducer for HttpProducer { - async fn start(&self) -> Result<(), Error> { - let producer = self.clone(); - let app = Router::new().route( - "/event", - post(move |event| { - let prod = producer.clone(); - async move { handle_event(event, prod).await } - }), - ); - - let addr = format!("0.0.0.0:{}", self.port); - let listener = tokio::net::TcpListener::bind(&addr).await?; - - let shutdown = self.shutdown.clone(); - tokio::select! { - result = axum::serve(listener, app) => { - result?; - Ok(()) - } - _ = shutdown.notified() => Ok(()) - } - } - - async fn stop(&self) -> Result<(), Error> { - self.shutdown.notify_one(); - Ok(()) - } - - async fn send_event(&self, event: Event) -> Result<(), Error> { - self.tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?; - Ok(()) - } - } - - async fn handle_event(Json(event): Json, producer: HttpProducer) -> Result<(), axum::http::StatusCode> { - producer - .send_event(event) - .await - .map_err(|_| axum::http::StatusCode::INTERNAL_SERVER_ERROR) - } -}