feat: implement event notification system

- Add core event notification interfaces
- Support multiple notification backends:
  - Webhook (default)
  - Kafka
  - MQTT
  - HTTP Producer
- Implement configurable event filtering
- Add async event dispatching with backpressure handling
- Provide serialization/deserialization for event payloads

This module enables system events to be published to various endpoints
with consistent delivery guarantees and failure handling.

crates/event-notifier/src/lib.rs

@@ -1,80 +1,131 @@
-/// RustFS Event Notifier
-/// This crate provides a simple event notification system for RustFS.
-/// It allows for the registration of event handlers and the triggering of events.
-///
+mod adapter;
+mod bus;
+mod config;
 mod error;
 mod event;
-mod event_name;
-mod notifier;
-mod rules;
-mod stats;
-mod target;
-mod target_entry;
+mod global;
+mod producer;
+mod store;
 
-pub fn add(left: u64, right: u64) -> u64 {
-    left + right
+pub use adapter::create_adapters;
+#[cfg(feature = "kafka")]
+pub use adapter::kafka::KafkaAdapter;
+#[cfg(feature = "mqtt")]
+pub use adapter::mqtt::MqttAdapter;
+#[cfg(feature = "webhook")]
+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")]
+pub use config::MqttConfig;
+#[cfg(feature = "webhook")]
+pub use config::WebhookConfig;
+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, send_event, shutdown, start};
+pub use store::EventStore;
+
+#[cfg(feature = "http-producer")]
+pub use producer::http::HttpProducer;
+#[cfg(feature = "http-producer")]
+pub use producer::EventProducer;
+
+use std::sync::Arc;
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+
+/// The `NotificationSystem` struct represents the notification system.
+/// It manages the event bus and the adapters.
+/// It is responsible for sending and receiving events.
+/// It also handles the shutdown process.
+pub struct NotificationSystem {
+    tx: mpsc::Sender<Event>,
+    rx: Option<mpsc::Receiver<Event>>,
+    store: Arc<EventStore>,
+    shutdown: CancellationToken,
+    #[cfg(feature = "http-producer")]
+    http_config: HttpProducerConfig,
 }
 
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::event::{Bucket, Event, Identity, Metadata, Object, Source};
-    use crate::target::{TargetID, TargetList};
-    use std::collections::HashMap;
+impl NotificationSystem {
+    /// Creates a new `NotificationSystem` instance.
+    pub async fn new(config: NotificationConfig) -> Result<Self, Error> {
+        let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity);
+        let store = Arc::new(EventStore::new(&config.store_path).await?);
+        let shutdown = CancellationToken::new();
 
-    #[test]
-    fn it_works() {
-        let result = add(2, 2);
-        assert_eq!(result, 4);
+        let restored_logs = store.load_logs().await?;
+        for log in restored_logs {
+            for event in log.records {
+                // For example, where the send method may return a SendError when calling it
+                tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?;
+            }
+        }
+
+        Ok(Self {
+            tx,
+            rx: Some(rx),
+            store,
+            shutdown,
+            #[cfg(feature = "http-producer")]
+            http_config: config.http,
+        })
     }
 
-    #[tokio::main]
-    async fn main() {
-        let target_list = TargetList::new();
-        let event = Event {
-            event_version: "1.0".to_string(),
-            event_source: "aws:s3".to_string(),
-            aws_region: "us-west-2".to_string(),
-            event_time: "2023-10-01T12:00:00Z".to_string(),
-            event_name: "PutObject".to_string(),
-            user_identity: Identity {
-                principal_id: "user123".to_string(),
+    /// Starts the notification system.
+    /// It initializes the event bus and the producer.
+    pub async fn start(&mut self, adapters: Vec<Arc<dyn ChannelAdapter>>) -> Result<(), Error> {
+        let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?;
+
+        let shutdown_clone = self.shutdown.clone();
+        let store_clone = self.store.clone();
+        let bus_handle = 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 = HttpProducer::new(self.tx.clone(), self.http_config.port);
+            producer.start().await?;
+        }
+
+        tokio::select! {
+            result = bus_handle => {
+                result.map_err(Error::JoinError)?;
+                Ok(())
             },
-            request_parameters: HashMap::new(),
-            response_elements: HashMap::new(),
-            s3: Metadata {
-                schema_version: "1.0".to_string(),
-                configuration_id: "config123".to_string(),
-                bucket: Bucket {
-                    name: "my-bucket".to_string(),
-                    owner_identity: Identity {
-                        principal_id: "owner123".to_string(),
-                    },
-                    arn: "arn:aws:s3:::my-bucket".to_string(),
-                },
-                object: Object {
-                    key: "my-object.txt".to_string(),
-                    version_id: Some("version123".to_string()),
-                    sequencer: "seq123".to_string(),
-                    size: Some(1024),
-                    etag: Some("etag123".to_string()),
-                    content_type: Some("text/plain".to_string()),
-                    user_metadata: HashMap::new(),
-                },
-            },
-            source: Source {
-                host: "localhost".to_string(),
-                user_agent: "RustFS/1.0".to_string(),
-            },
-        };
-        let target_ids: &[TargetID] = &["".to_string()];
-        // 发送事件
-        let results = target_list.send(event, &target_ids).await;
-        println!("result len:{:?}", results.len());
-        // 获取统计信息
-        let stats = target_list.get_stats();
-        for (id, stat) in stats {
-            println!("Target {}: {} events, {} failed", id, stat.total_events, stat.failed_events);
+            _ = self.shutdown.cancelled() => {
+                tracing::info!("System shutdown triggered");
+                Ok(())
+            }
         }
     }
+
+    /// Sends an event to the notification system.
+    /// This method is used to send events to the event bus.
+    pub async fn send_event(&self, event: Event) -> Result<(), Error> {
+        self.tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?;
+        Ok(())
+    }
+
+    /// Shuts down the notification system.
+    /// This method is used to cancel the event bus and producer tasks.
+    pub fn shutdown(&self) {
+        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;
+    }
 }