improve code

Cargo.lock

@@ -7350,7 +7350,6 @@ name = "rustfs-event-notifier"
 version = "0.0.1"
 dependencies = [
  "async-trait",
- "axum",
  "dotenvy",
  "figment",
  "rdkafka",

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 }

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<Vec<Arc<dyn ChannelAdapter>>, Box<Error>> {
+pub fn create_adapters(configs: &[AdapterConfig]) -> Result<Vec<Arc<dyn ChannelAdapter>>, Error> {
     let mut adapters: Vec<Arc<dyn ChannelAdapter>> = 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<Vec<Arc<dyn ChannelA
                 adapters.push(Arc::new(mqtt));
             }
             #[cfg(not(feature = "webhook"))]
-            AdapterConfig::Webhook(_) => 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")),
         }
     }
 

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<AdapterConfig>,
-    #[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
-}

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),

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<Arc<Mutex<NotificationSystem>>> = 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<Arc<dyn ChannelAdapter>>) -> 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<Arc<dyn ChannelAdapter>>) -> 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<Arc<Mutex<NotificationSystem>>, 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");
+    }
 }

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;

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<mpsc::Receiver<Event>>,
     store: Arc<EventStore>,
     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;
-    }
 }

crates/event-notifier/src/producer.rs

@@ -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<Event>,
-        port: u16,
-        shutdown: Arc<tokio::sync::Notify>,
-    }
-
-    impl HttpProducer {
-        pub fn new(tx: mpsc::Sender<Event>, 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<Event>, producer: HttpProducer) -> Result<(), axum::http::StatusCode> {
-        producer
-            .send_event(event)
-            .await
-            .map_err(|_| axum::http::StatusCode::INTERNAL_SERVER_ERROR)
-    }
-}