init event crate

Cargo.lock

@@ -8126,6 +8126,27 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "rustfs-event"
+version = "0.0.1"
+dependencies = [
+ "common",
+ "ecstore",
+ "once_cell",
+ "reqwest",
+ "rustfs-config",
+ "serde",
+ "serde_json",
+ "serde_with",
+ "smallvec",
+ "strum",
+ "thiserror 2.0.12",
+ "tokio",
+ "tokio-util",
+ "tracing",
+ "uuid",
+]
+
 [[package]]
 name = "rustfs-gui"
 version = "0.0.1"
@@ -8156,11 +8177,10 @@ dependencies = [
  "common",
  "dotenvy",
  "ecstore",
- "lazy_static",
  "once_cell",
- "rdkafka",
  "reqwest",
  "rumqttc",
+ "rustfs-config",
  "serde",
  "serde_json",
  "serde_with",
@@ -9047,9 +9067,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.15.0"
+version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9"
+checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
 dependencies = [
  "serde",
 ]

Cargo.toml

@@ -7,9 +7,11 @@ members = [
     "common/protos", # Protocol buffer definitions
     "common/workers", # Worker thread pools and task scheduling
     "crates/config", # Configuration management
-    "crates/notify", # Event notification system
+    "crates/event", # Event handling and processing
+    "crates/notify", # Notification system for events
     "crates/obs", # Observability utilities
     "crates/utils", # Utility functions and helpers
+    "crates/zip", # ZIP file handling and compression
     "crypto", # Cryptography and security features
     "ecstore", # Erasure coding storage implementation
     "e2e_test", # End-to-end test suite
@@ -18,7 +20,6 @@ members = [
     "rustfs", # Core file system implementation
     "s3select/api", # S3 Select API interface
     "s3select/query", # S3 Select query engine
-    "crates/zip",
 ]
 resolver = "2"
 
@@ -74,7 +75,7 @@ byteorder = "1.5.0"
 cfg-if = "1.0.0"
 chacha20poly1305 = { version = "0.10.1" }
 chrono = { version = "0.4.41", features = ["serde"] }
-clap = { version = "4.5.39", features = ["derive", "env"] }
+clap = { version = "4.5.40", features = ["derive", "env"] }
 const-str = { version = "0.6.2", features = ["std", "proc"] }
 crc32fast = "1.4.2"
 datafusion = "46.0.1"
@@ -184,7 +185,7 @@ serde_urlencoded = "0.7.1"
 serde_with = "3.12.0"
 sha2 = "0.10.9"
 siphasher = "1.0.1"
-smallvec = { version = "1.15.0", features = ["serde"] }
+smallvec = { version = "1.15.1", features = ["serde"] }
 
 snafu = "0.8.6"
 snap = "1.1.1"

crates/config/src/notify/webhook.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 
 /// WebhookArgs - Webhook target arguments.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -7,6 +8,7 @@ pub struct WebhookArgs {
     pub endpoint: String,
     pub auth_token: String,
     #[serde(skip)]
+    pub custom_headers: Option<HashMap<String, String>>,
     pub queue_dir: String,
     pub queue_limit: u64,
     pub client_cert: String,
@@ -20,6 +22,7 @@ impl WebhookArgs {
             enable: false,
             endpoint: "".to_string(),
             auth_token: "".to_string(),
+            custom_headers: None,
             queue_dir: "".to_string(),
             queue_limit: 0,
             client_cert: "".to_string(),
@@ -53,13 +56,14 @@ impl Default for WebhookArgs {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
+    use crate::notify::webhook::WebhookArgs;
 
     #[test]
     fn test_webhook_args_new() {
         let args = WebhookArgs::new();
         assert_eq!(args.endpoint, "");
         assert_eq!(args.auth_token, "");
+        assert!(args.custom_headers.is_none());
         assert_eq!(args.queue_dir, "");
         assert_eq!(args.queue_limit, 0);
         assert_eq!(args.client_cert, "");

crates/event/Cargo.toml

@@ -0,0 +1,27 @@
+[package]
+name = "rustfs-event"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+rustfs-config = { workspace = true, features = ["constants", "notify"] }
+common = { workspace = true }
+ecstore = { workspace = true }
+once_cell = { workspace = true }
+reqwest = { workspace = true, optional = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+serde_with = { workspace = true }
+smallvec = { workspace = true, features = ["serde"] }
+strum = { workspace = true, features = ["derive"] }
+tracing = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["sync", "net", "macros", "signal", "rt-multi-thread"] }
+tokio-util = { workspace = true }
+uuid = { workspace = true, features = ["v4", "serde"] }
+
+[lints]
+workspace = true

crates/event/src/error.rs

@@ -0,0 +1,403 @@
+use thiserror::Error;
+use tokio::sync::mpsc::error;
+use tokio::task::JoinError;
+
+/// The `Error` enum represents all possible errors that can occur in the application.
+/// It implements the `std::error::Error` trait and provides a way to convert various error types into a single error type.
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("Join error: {0}")]
+    JoinError(#[from] JoinError),
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("Serialization error: {0}")]
+    Serde(#[from] serde_json::Error),
+    #[error("HTTP error: {0}")]
+    Http(#[from] reqwest::Error),
+    #[cfg(all(feature = "kafka", target_os = "linux"))]
+    #[error("Kafka error: {0}")]
+    Kafka(#[from] rdkafka::error::KafkaError),
+    #[cfg(feature = "mqtt")]
+    #[error("MQTT error: {0}")]
+    Mqtt(#[from] rumqttc::ClientError),
+    #[error("Channel send error: {0}")]
+    ChannelSend(#[from] Box<error::SendError<crate::event::Event>>),
+    #[error("Feature disabled: {0}")]
+    FeatureDisabled(&'static str),
+    #[error("Event bus already started")]
+    EventBusStarted,
+    #[error("necessary fields are missing:{0}")]
+    MissingField(&'static str),
+    #[error("field verification failed:{0}")]
+    ValidationError(&'static str),
+    #[error("Custom error: {0}")]
+    Custom(String),
+    #[error("Configuration error: {0}")]
+    ConfigError(String),
+    #[error("create adapter failed error: {0}")]
+    AdapterCreationFailed(String),
+}
+
+impl Error {
+    pub fn custom(msg: &str) -> Error {
+        Self::Custom(msg.to_string())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::error::Error as StdError;
+    use std::io;
+    use tokio::sync::mpsc;
+
+    #[test]
+    fn test_error_display() {
+        // Test error message display
+        let custom_error = Error::custom("test message");
+        assert_eq!(custom_error.to_string(), "Custom error: test message");
+
+        let feature_error = Error::FeatureDisabled("test feature");
+        assert_eq!(feature_error.to_string(), "Feature disabled: test feature");
+
+        let event_bus_error = Error::EventBusStarted;
+        assert_eq!(event_bus_error.to_string(), "Event bus already started");
+
+        let missing_field_error = Error::MissingField("required_field");
+        assert_eq!(missing_field_error.to_string(), "necessary fields are missing:required_field");
+
+        let validation_error = Error::ValidationError("invalid format");
+        assert_eq!(validation_error.to_string(), "field verification failed:invalid format");
+
+        let config_error = Error::ConfigError("invalid config".to_string());
+        assert_eq!(config_error.to_string(), "Configuration error: invalid config");
+    }
+
+    #[test]
+    fn test_error_debug() {
+        // Test Debug trait implementation
+        let custom_error = Error::custom("debug test");
+        let debug_str = format!("{:?}", custom_error);
+        assert!(debug_str.contains("Custom"));
+        assert!(debug_str.contains("debug test"));
+
+        let feature_error = Error::FeatureDisabled("debug feature");
+        let debug_str = format!("{:?}", feature_error);
+        assert!(debug_str.contains("FeatureDisabled"));
+        assert!(debug_str.contains("debug feature"));
+    }
+
+    #[test]
+    fn test_custom_error_creation() {
+        // Test custom error creation
+        let error = Error::custom("test custom error");
+        match error {
+            Error::Custom(msg) => assert_eq!(msg, "test custom error"),
+            _ => panic!("Expected Custom error variant"),
+        }
+
+        // Test empty string
+        let empty_error = Error::custom("");
+        match empty_error {
+            Error::Custom(msg) => assert_eq!(msg, ""),
+            _ => panic!("Expected Custom error variant"),
+        }
+
+        // Test special characters
+        let special_error = Error::custom("Test Chinese 中文 & special chars: !@#$%");
+        match special_error {
+            Error::Custom(msg) => assert_eq!(msg, "Test Chinese 中文 & special chars: !@#$%"),
+            _ => panic!("Expected Custom error variant"),
+        }
+    }
+
+    #[test]
+    fn test_io_error_conversion() {
+        // Test IO error conversion
+        let io_error = io::Error::new(io::ErrorKind::NotFound, "file not found");
+        let converted_error: Error = io_error.into();
+
+        match converted_error {
+            Error::Io(err) => {
+                assert_eq!(err.kind(), io::ErrorKind::NotFound);
+                assert_eq!(err.to_string(), "file not found");
+            }
+            _ => panic!("Expected Io error variant"),
+        }
+
+        // Test different types of IO errors
+        let permission_error = io::Error::new(io::ErrorKind::PermissionDenied, "access denied");
+        let converted: Error = permission_error.into();
+        assert!(matches!(converted, Error::Io(_)));
+    }
+
+    #[test]
+    fn test_serde_error_conversion() {
+        // Test serialization error conversion
+        let invalid_json = r#"{"invalid": json}"#;
+        let serde_error = serde_json::from_str::<serde_json::Value>(invalid_json).unwrap_err();
+        let converted_error: Error = serde_error.into();
+
+        match converted_error {
+            Error::Serde(_) => {
+                // Verify error type is correct
+                assert!(converted_error.to_string().contains("Serialization error"));
+            }
+            _ => panic!("Expected Serde error variant"),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_channel_send_error_conversion() {
+        // Test channel send error conversion
+        let (tx, rx) = mpsc::channel::<crate::event::Event>(1);
+        drop(rx); // Close receiver
+
+        // Create a test event
+        use crate::event::{Bucket, Identity, Metadata, Name, Object, Source};
+        use std::collections::HashMap;
+
+        let identity = Identity::new("test-user".to_string());
+        let bucket = Bucket::new("test-bucket".to_string(), identity.clone(), "arn:aws:s3:::test-bucket".to_string());
+        let object = Object::new(
+            "test-key".to_string(),
+            Some(1024),
+            Some("etag123".to_string()),
+            Some("text/plain".to_string()),
+            Some(HashMap::new()),
+            None,
+            "sequencer123".to_string(),
+        );
+        let metadata = Metadata::create("1.0".to_string(), "config1".to_string(), bucket, object);
+        let source = Source::new("localhost".to_string(), "8080".to_string(), "test-agent".to_string());
+
+        let test_event = crate::event::Event::builder()
+            .event_name(Name::ObjectCreatedPut)
+            .s3(metadata)
+            .source(source)
+            .build()
+            .unwrap();
+
+        let send_result = tx.send(test_event).await;
+        assert!(send_result.is_err());
+
+        let send_error = send_result.unwrap_err();
+        let boxed_error = Box::new(send_error);
+        let converted_error: Error = boxed_error.into();
+
+        match converted_error {
+            Error::ChannelSend(_) => {
+                assert!(converted_error.to_string().contains("Channel send error"));
+            }
+            _ => panic!("Expected ChannelSend error variant"),
+        }
+    }
+
+    #[test]
+    fn test_error_source_chain() {
+        // 测试错误源链
+        let io_error = io::Error::new(io::ErrorKind::InvalidData, "invalid data");
+        let converted_error: Error = io_error.into();
+
+        // 验证错误源
+        assert!(converted_error.source().is_some());
+        let source = converted_error.source().unwrap();
+        assert_eq!(source.to_string(), "invalid data");
+    }
+
+    #[test]
+    fn test_error_variants_exhaustive() {
+        // 测试所有错误变体的创建
+        let errors = vec![
+            Error::FeatureDisabled("test"),
+            Error::EventBusStarted,
+            Error::MissingField("field"),
+            Error::ValidationError("validation"),
+            Error::Custom("custom".to_string()),
+            Error::ConfigError("config".to_string()),
+        ];
+
+        for error in errors {
+            // 验证每个错误都能正确显示
+            let error_str = error.to_string();
+            assert!(!error_str.is_empty());
+
+            // 验证每个错误都能正确调试
+            let debug_str = format!("{:?}", error);
+            assert!(!debug_str.is_empty());
+        }
+    }
+
+    #[test]
+    fn test_error_equality_and_matching() {
+        // 测试错误的模式匹配
+        let custom_error = Error::custom("test");
+        match custom_error {
+            Error::Custom(msg) => assert_eq!(msg, "test"),
+            _ => panic!("Pattern matching failed"),
+        }
+
+        let feature_error = Error::FeatureDisabled("feature");
+        match feature_error {
+            Error::FeatureDisabled(feature) => assert_eq!(feature, "feature"),
+            _ => panic!("Pattern matching failed"),
+        }
+
+        let event_bus_error = Error::EventBusStarted;
+        match event_bus_error {
+            Error::EventBusStarted => {} // 正确匹配
+            _ => panic!("Pattern matching failed"),
+        }
+    }
+
+    #[test]
+    fn test_error_message_formatting() {
+        // 测试错误消息格式化
+        let test_cases = vec![
+            (Error::FeatureDisabled("kafka"), "Feature disabled: kafka"),
+            (Error::MissingField("bucket_name"), "necessary fields are missing:bucket_name"),
+            (Error::ValidationError("invalid email"), "field verification failed:invalid email"),
+            (Error::ConfigError("missing file".to_string()), "Configuration error: missing file"),
+        ];
+
+        for (error, expected_message) in test_cases {
+            assert_eq!(error.to_string(), expected_message);
+        }
+    }
+
+    #[test]
+    fn test_error_memory_efficiency() {
+        // 测试错误类型的内存效率
+        use std::mem;
+
+        let size = mem::size_of::<Error>();
+        // 错误类型应该相对紧凑，考虑到包含多种错误类型，96 字节是合理的
+        assert!(size <= 128, "Error size should be reasonable, got {} bytes", size);
+
+        // 测试 Option<Error>的大小
+        let option_size = mem::size_of::<Option<Error>>();
+        assert!(option_size <= 136, "Option<Error> should be efficient, got {} bytes", option_size);
+    }
+
+    #[test]
+    fn test_error_thread_safety() {
+        // 测试错误类型的线程安全性
+        fn assert_send<T: Send>() {}
+        fn assert_sync<T: Sync>() {}
+
+        assert_send::<Error>();
+        assert_sync::<Error>();
+    }
+
+    #[test]
+    fn test_custom_error_edge_cases() {
+        // 测试自定义错误的边界情况
+        let long_message = "a".repeat(1000);
+        let long_error = Error::custom(&long_message);
+        match long_error {
+            Error::Custom(msg) => assert_eq!(msg.len(), 1000),
+            _ => panic!("Expected Custom error variant"),
+        }
+
+        // 测试包含换行符的消息
+        let multiline_error = Error::custom("line1\nline2\nline3");
+        match multiline_error {
+            Error::Custom(msg) => assert!(msg.contains('\n')),
+            _ => panic!("Expected Custom error variant"),
+        }
+
+        // 测试包含 Unicode 字符的消息
+        let unicode_error = Error::custom("🚀 Unicode test 测试 🎉");
+        match unicode_error {
+            Error::Custom(msg) => assert!(msg.contains('🚀')),
+            _ => panic!("Expected Custom error variant"),
+        }
+    }
+
+    #[test]
+    fn test_error_conversion_consistency() {
+        // 测试错误转换的一致性
+        let original_io_error = io::Error::new(io::ErrorKind::TimedOut, "timeout");
+        let error_message = original_io_error.to_string();
+        let converted: Error = original_io_error.into();
+
+        // 验证转换后的错误包含原始错误信息
+        assert!(converted.to_string().contains(&error_message));
+    }
+
+    #[test]
+    fn test_error_downcast() {
+        // 测试错误的向下转型
+        let io_error = io::Error::other("test error");
+        let converted: Error = io_error.into();
+
+        // 验证可以获取源错误
+        if let Error::Io(ref inner) = converted {
+            assert_eq!(inner.to_string(), "test error");
+            assert_eq!(inner.kind(), io::ErrorKind::Other);
+        } else {
+            panic!("Expected Io error variant");
+        }
+    }
+
+    #[test]
+    fn test_error_chain_depth() {
+        // 测试错误链的深度
+        let root_cause = io::Error::other("root cause");
+        let converted: Error = root_cause.into();
+
+        let mut depth = 0;
+        let mut current_error: &dyn StdError = &converted;
+
+        while let Some(source) = current_error.source() {
+            depth += 1;
+            current_error = source;
+            // 防止无限循环
+            if depth > 10 {
+                break;
+            }
+        }
+
+        assert!(depth > 0, "Error should have at least one source");
+        assert!(depth <= 3, "Error chain should not be too deep");
+    }
+
+    #[test]
+    fn test_static_str_lifetime() {
+        // 测试静态字符串生命周期
+        fn create_feature_error() -> Error {
+            Error::FeatureDisabled("static_feature")
+        }
+
+        let error = create_feature_error();
+        match error {
+            Error::FeatureDisabled(feature) => assert_eq!(feature, "static_feature"),
+            _ => panic!("Expected FeatureDisabled error variant"),
+        }
+    }
+
+    #[test]
+    fn test_error_formatting_consistency() {
+        // 测试错误格式化的一致性
+        let errors = vec![
+            Error::FeatureDisabled("test"),
+            Error::MissingField("field"),
+            Error::ValidationError("validation"),
+            Error::Custom("custom".to_string()),
+        ];
+
+        for error in errors {
+            let display_str = error.to_string();
+            let debug_str = format!("{:?}", error);
+
+            // Display 和 Debug 都不应该为空
+            assert!(!display_str.is_empty());
+            assert!(!debug_str.is_empty());
+
+            // Debug 输出通常包含更多信息，但不是绝对的
+            // 这里我们只验证两者都有内容即可
+            assert!(!debug_str.is_empty());
+            assert!(!display_str.is_empty());
+        }
+    }
+}

crates/event/src/event.rs

@@ -0,0 +1,617 @@
+use crate::Error;
+use reqwest::dns::Name;
+use serde::{Deserialize, Serialize};
+use serde_with::{DeserializeFromStr, SerializeDisplay};
+use smallvec::{smallvec, SmallVec};
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::time::{SystemTime, UNIX_EPOCH};
+use strum::{Display, EnumString};
+use uuid::Uuid;
+
+/// A struct representing the identity of the user
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Identity {
+    #[serde(rename = "principalId")]
+    pub principal_id: String,
+}
+
+impl Identity {
+    /// Create a new Identity instance
+    pub fn new(principal_id: String) -> Self {
+        Self { principal_id }
+    }
+
+    /// Set the principal ID
+    pub fn set_principal_id(&mut self, principal_id: String) {
+        self.principal_id = principal_id;
+    }
+}
+
+/// A struct representing the bucket information
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Bucket {
+    pub name: String,
+    #[serde(rename = "ownerIdentity")]
+    pub owner_identity: Identity,
+    pub arn: String,
+}
+
+impl Bucket {
+    /// Create a new Bucket instance
+    pub fn new(name: String, owner_identity: Identity, arn: String) -> Self {
+        Self {
+            name,
+            owner_identity,
+            arn,
+        }
+    }
+
+    /// Set the name of the bucket
+    pub fn set_name(&mut self, name: String) {
+        self.name = name;
+    }
+
+    /// Set the ARN of the bucket
+    pub fn set_arn(&mut self, arn: String) {
+        self.arn = arn;
+    }
+
+    /// Set the owner identity of the bucket
+    pub fn set_owner_identity(&mut self, owner_identity: Identity) {
+        self.owner_identity = owner_identity;
+    }
+}
+
+/// A struct representing the object information
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Object {
+    pub key: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub size: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none", rename = "eTag")]
+    pub etag: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none", rename = "contentType")]
+    pub content_type: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none", rename = "userMetadata")]
+    pub user_metadata: Option<HashMap<String, String>>,
+    #[serde(default, skip_serializing_if = "Option::is_none", rename = "versionId")]
+    pub version_id: Option<String>,
+    pub sequencer: String,
+}
+
+impl Object {
+    /// Create a new Object instance
+    pub fn new(
+        key: String,
+        size: Option<i64>,
+        etag: Option<String>,
+        content_type: Option<String>,
+        user_metadata: Option<HashMap<String, String>>,
+        version_id: Option<String>,
+        sequencer: String,
+    ) -> Self {
+        Self {
+            key,
+            size,
+            etag,
+            content_type,
+            user_metadata,
+            version_id,
+            sequencer,
+        }
+    }
+
+    /// Set the key
+    pub fn set_key(&mut self, key: String) {
+        self.key = key;
+    }
+
+    /// Set the size
+    pub fn set_size(&mut self, size: Option<i64>) {
+        self.size = size;
+    }
+
+    /// Set the etag
+    pub fn set_etag(&mut self, etag: Option<String>) {
+        self.etag = etag;
+    }
+
+    /// Set the content type
+    pub fn set_content_type(&mut self, content_type: Option<String>) {
+        self.content_type = content_type;
+    }
+
+    /// Set the user metadata
+    pub fn set_user_metadata(&mut self, user_metadata: Option<HashMap<String, String>>) {
+        self.user_metadata = user_metadata;
+    }
+
+    /// Set the version ID
+    pub fn set_version_id(&mut self, version_id: Option<String>) {
+        self.version_id = version_id;
+    }
+
+    /// Set the sequencer
+    pub fn set_sequencer(&mut self, sequencer: String) {
+        self.sequencer = sequencer;
+    }
+}
+
+/// A struct representing the metadata of the event
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Metadata {
+    #[serde(rename = "s3SchemaVersion")]
+    pub schema_version: String,
+    #[serde(rename = "configurationId")]
+    pub configuration_id: String,
+    pub bucket: Bucket,
+    pub object: Object,
+}
+
+impl Default for Metadata {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+impl Metadata {
+    /// Create a new Metadata instance with default values
+    pub fn new() -> Self {
+        Self {
+            schema_version: "1.0".to_string(),
+            configuration_id: "default".to_string(),
+            bucket: Bucket::new(
+                "default".to_string(),
+                Identity::new("default".to_string()),
+                "arn:aws:s3:::default".to_string(),
+            ),
+            object: Object::new("default".to_string(), None, None, None, None, None, "default".to_string()),
+        }
+    }
+
+    /// Create a new Metadata instance
+    pub fn create(schema_version: String, configuration_id: String, bucket: Bucket, object: Object) -> Self {
+        Self {
+            schema_version,
+            configuration_id,
+            bucket,
+            object,
+        }
+    }
+
+    /// Set the schema version
+    pub fn set_schema_version(&mut self, schema_version: String) {
+        self.schema_version = schema_version;
+    }
+
+    /// Set the configuration ID
+    pub fn set_configuration_id(&mut self, configuration_id: String) {
+        self.configuration_id = configuration_id;
+    }
+
+    /// Set the bucket
+    pub fn set_bucket(&mut self, bucket: Bucket) {
+        self.bucket = bucket;
+    }
+
+    /// Set the object
+    pub fn set_object(&mut self, object: Object) {
+        self.object = object;
+    }
+}
+
+/// A struct representing the source of the event
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Source {
+    pub host: String,
+    pub port: String,
+    #[serde(rename = "userAgent")]
+    pub user_agent: String,
+}
+
+impl Source {
+    /// Create a new Source instance
+    pub fn new(host: String, port: String, user_agent: String) -> Self {
+        Self { host, port, user_agent }
+    }
+
+    /// Set the host
+    pub fn set_host(&mut self, host: String) {
+        self.host = host;
+    }
+
+    /// Set the port
+    pub fn set_port(&mut self, port: String) {
+        self.port = port;
+    }
+
+    /// Set the user agent
+    pub fn set_user_agent(&mut self, user_agent: String) {
+        self.user_agent = user_agent;
+    }
+}
+
+/// Builder for creating an Event.
+///
+/// This struct is used to build an Event object with various parameters.
+/// It provides methods to set each parameter and a build method to create the Event.
+#[derive(Default, Clone)]
+pub struct EventBuilder {
+    event_version: Option<String>,
+    event_source: Option<String>,
+    aws_region: Option<String>,
+    event_time: Option<String>,
+    event_name: Option<Name>,
+    user_identity: Option<Identity>,
+    request_parameters: Option<HashMap<String, String>>,
+    response_elements: Option<HashMap<String, String>>,
+    s3: Option<Metadata>,
+    source: Option<Source>,
+    channels: Option<SmallVec<[String; 2]>>,
+}
+
+impl EventBuilder {
+    /// create a builder that pre filled default values
+    pub fn new() -> Self {
+        Self {
+            event_version: Some(Cow::Borrowed("2.0").to_string()),
+            event_source: Some(Cow::Borrowed("aws:s3").to_string()),
+            aws_region: Some("us-east-1".to_string()),
+            event_time: Some(SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string()),
+            event_name: None,
+            user_identity: Some(Identity {
+                principal_id: "anonymous".to_string(),
+            }),
+            request_parameters: Some(HashMap::new()),
+            response_elements: Some(HashMap::new()),
+            s3: None,
+            source: None,
+            channels: Some(Vec::new().into()),
+        }
+    }
+
+    /// verify and set the event version
+    pub fn event_version(mut self, event_version: impl Into<String>) -> Self {
+        let event_version = event_version.into();
+        if !event_version.is_empty() {
+            self.event_version = Some(event_version);
+        }
+        self
+    }
+
+    /// verify and set the event source
+    pub fn event_source(mut self, event_source: impl Into<String>) -> Self {
+        let event_source = event_source.into();
+        if !event_source.is_empty() {
+            self.event_source = Some(event_source);
+        }
+        self
+    }
+
+    /// set up aws regions
+    pub fn aws_region(mut self, aws_region: impl Into<String>) -> Self {
+        self.aws_region = Some(aws_region.into());
+        self
+    }
+
+    /// set event time
+    pub fn event_time(mut self, event_time: impl Into<String>) -> Self {
+        self.event_time = Some(event_time.into());
+        self
+    }
+
+    /// set event name
+    pub fn event_name(mut self, event_name: Name) -> Self {
+        self.event_name = Some(event_name);
+        self
+    }
+
+    /// set user identity
+    pub fn user_identity(mut self, user_identity: Identity) -> Self {
+        self.user_identity = Some(user_identity);
+        self
+    }
+
+    /// set request parameters
+    pub fn request_parameters(mut self, request_parameters: HashMap<String, String>) -> Self {
+        self.request_parameters = Some(request_parameters);
+        self
+    }
+
+    /// set response elements
+    pub fn response_elements(mut self, response_elements: HashMap<String, String>) -> Self {
+        self.response_elements = Some(response_elements);
+        self
+    }
+
+    /// setting up s3 metadata
+    pub fn s3(mut self, s3: Metadata) -> Self {
+        self.s3 = Some(s3);
+        self
+    }
+
+    /// set event source information
+    pub fn source(mut self, source: Source) -> Self {
+        self.source = Some(source);
+        self
+    }
+
+    /// set up the sending channel
+    pub fn channels(mut self, channels: Vec<String>) -> Self {
+        self.channels = Some(channels.into());
+        self
+    }
+
+    /// Create a preconfigured builder for common object event scenarios
+    pub fn for_object_creation(s3: Metadata, source: Source) -> Self {
+        Self::new().event_name(Name::ObjectCreatedPut).s3(s3).source(source)
+    }
+
+    /// Create a preconfigured builder for object deletion events
+    pub fn for_object_removal(s3: Metadata, source: Source) -> Self {
+        Self::new().event_name(Name::ObjectRemovedDelete).s3(s3).source(source)
+    }
+
+    /// build event instance
+    ///
+    /// Verify the required fields and create a complete Event object
+    pub fn build(self) -> Result<Event, Error> {
+        let event_version = self.event_version.ok_or(Error::MissingField("event_version"))?;
+
+        let event_source = self.event_source.ok_or(Error::MissingField("event_source"))?;
+
+        let aws_region = self.aws_region.ok_or(Error::MissingField("aws_region"))?;
+
+        let event_time = self.event_time.ok_or(Error::MissingField("event_time"))?;
+
+        let event_name = self.event_name.ok_or(Error::MissingField("event_name"))?;
+
+        let user_identity = self.user_identity.ok_or(Error::MissingField("user_identity"))?;
+
+        let request_parameters = self.request_parameters.unwrap_or_default();
+        let response_elements = self.response_elements.unwrap_or_default();
+
+        let s3 = self.s3.ok_or(Error::MissingField("s3"))?;
+
+        let source = self.source.ok_or(Error::MissingField("source"))?;
+
+        let channels = self.channels.unwrap_or_else(|| smallvec![]);
+
+        Ok(Event {
+            event_version,
+            event_source,
+            aws_region,
+            event_time,
+            event_name,
+            user_identity,
+            request_parameters,
+            response_elements,
+            s3,
+            source,
+            id: Uuid::new_v4(),
+            timestamp: SystemTime::now(),
+            channels,
+        })
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Event {
+    #[serde(rename = "eventVersion")]
+    pub event_version: String,
+    #[serde(rename = "eventSource")]
+    pub event_source: String,
+    #[serde(rename = "awsRegion")]
+    pub aws_region: String,
+    #[serde(rename = "eventTime")]
+    pub event_time: String,
+    #[serde(rename = "eventName")]
+    pub event_name: Name,
+    #[serde(rename = "userIdentity")]
+    pub user_identity: Identity,
+    #[serde(rename = "requestParameters")]
+    pub request_parameters: HashMap<String, String>,
+    #[serde(rename = "responseElements")]
+    pub response_elements: HashMap<String, String>,
+    pub s3: Metadata,
+    pub source: Source,
+    pub id: Uuid,
+    pub timestamp: SystemTime,
+    pub channels: SmallVec<[String; 2]>,
+}
+
+impl Event {
+    /// create a new event builder
+    ///
+    /// Returns an EventBuilder instance pre-filled with default values
+    pub fn builder() -> EventBuilder {
+        EventBuilder::new()
+    }
+
+    /// Quickly create Event instances with necessary fields
+    ///
+    /// suitable for common s3 event scenarios
+    pub fn create(event_name: Name, s3: Metadata, source: Source, channels: Vec<String>) -> Self {
+        Self::builder()
+            .event_name(event_name)
+            .s3(s3)
+            .source(source)
+            .channels(channels)
+            .build()
+            .expect("Failed to create event, missing necessary parameters")
+    }
+
+    /// a convenient way to create a preconfigured builder
+    pub fn for_object_creation(s3: Metadata, source: Source) -> EventBuilder {
+        EventBuilder::for_object_creation(s3, source)
+    }
+
+    /// a convenient way to create a preconfigured builder
+    pub fn for_object_removal(s3: Metadata, source: Source) -> EventBuilder {
+        EventBuilder::for_object_removal(s3, source)
+    }
+
+    /// Determine whether an event belongs to a specific type
+    pub fn is_type(&self, event_type: Name) -> bool {
+        let mask = event_type.mask();
+        (self.event_name.mask() & mask) != 0
+    }
+
+    /// Determine whether an event needs to be sent to a specific channel
+    pub fn is_for_channel(&self, channel: &str) -> bool {
+        self.channels.iter().any(|c| c == channel)
+    }
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Log {
+    #[serde(rename = "eventName")]
+    pub event_name: Name,
+    pub key: String,
+    pub records: Vec<Event>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerializeDisplay, DeserializeFromStr, Display, EnumString)]
+#[strum(serialize_all = "SCREAMING_SNAKE_CASE")]
+pub enum Name {
+    ObjectAccessedGet,
+    ObjectAccessedGetRetention,
+    ObjectAccessedGetLegalHold,
+    ObjectAccessedHead,
+    ObjectAccessedAttributes,
+    ObjectCreatedCompleteMultipartUpload,
+    ObjectCreatedCopy,
+    ObjectCreatedPost,
+    ObjectCreatedPut,
+    ObjectCreatedPutRetention,
+    ObjectCreatedPutLegalHold,
+    ObjectCreatedPutTagging,
+    ObjectCreatedDeleteTagging,
+    ObjectRemovedDelete,
+    ObjectRemovedDeleteMarkerCreated,
+    ObjectRemovedDeleteAllVersions,
+    ObjectRemovedNoOp,
+    BucketCreated,
+    BucketRemoved,
+    ObjectReplicationFailed,
+    ObjectReplicationComplete,
+    ObjectReplicationMissedThreshold,
+    ObjectReplicationReplicatedAfterThreshold,
+    ObjectReplicationNotTracked,
+    ObjectRestorePost,
+    ObjectRestoreCompleted,
+    ObjectTransitionFailed,
+    ObjectTransitionComplete,
+    ObjectManyVersions,
+    ObjectLargeVersions,
+    PrefixManyFolders,
+    IlmDelMarkerExpirationDelete,
+    ObjectAccessedAll,
+    ObjectCreatedAll,
+    ObjectRemovedAll,
+    ObjectReplicationAll,
+    ObjectRestoreAll,
+    ObjectTransitionAll,
+    ObjectScannerAll,
+    Everything,
+}
+
+impl Name {
+    pub fn expand(&self) -> Vec<Name> {
+        match self {
+            Name::ObjectAccessedAll => vec![
+                Name::ObjectAccessedGet,
+                Name::ObjectAccessedHead,
+                Name::ObjectAccessedGetRetention,
+                Name::ObjectAccessedGetLegalHold,
+                Name::ObjectAccessedAttributes,
+            ],
+            Name::ObjectCreatedAll => vec![
+                Name::ObjectCreatedCompleteMultipartUpload,
+                Name::ObjectCreatedCopy,
+                Name::ObjectCreatedPost,
+                Name::ObjectCreatedPut,
+                Name::ObjectCreatedPutRetention,
+                Name::ObjectCreatedPutLegalHold,
+                Name::ObjectCreatedPutTagging,
+                Name::ObjectCreatedDeleteTagging,
+            ],
+            Name::ObjectRemovedAll => vec![
+                Name::ObjectRemovedDelete,
+                Name::ObjectRemovedDeleteMarkerCreated,
+                Name::ObjectRemovedNoOp,
+                Name::ObjectRemovedDeleteAllVersions,
+            ],
+            Name::ObjectReplicationAll => vec![
+                Name::ObjectReplicationFailed,
+                Name::ObjectReplicationComplete,
+                Name::ObjectReplicationNotTracked,
+                Name::ObjectReplicationMissedThreshold,
+                Name::ObjectReplicationReplicatedAfterThreshold,
+            ],
+            Name::ObjectRestoreAll => vec![Name::ObjectRestorePost, Name::ObjectRestoreCompleted],
+            Name::ObjectTransitionAll => {
+                vec![Name::ObjectTransitionFailed, Name::ObjectTransitionComplete]
+            }
+            Name::ObjectScannerAll => vec![Name::ObjectManyVersions, Name::ObjectLargeVersions, Name::PrefixManyFolders],
+            Name::Everything => (1..=Name::IlmDelMarkerExpirationDelete as u32)
+                .map(|i| Name::from_repr(i).unwrap())
+                .collect(),
+            _ => vec![*self],
+        }
+    }
+
+    pub fn mask(&self) -> u64 {
+        if (*self as u32) < Name::ObjectAccessedAll as u32 {
+            1 << (*self as u32 - 1)
+        } else {
+            self.expand().iter().fold(0, |acc, n| acc | (1 << (*n as u32 - 1)))
+        }
+    }
+
+    fn from_repr(discriminant: u32) -> Option<Self> {
+        match discriminant {
+            1 => Some(Name::ObjectAccessedGet),
+            2 => Some(Name::ObjectAccessedGetRetention),
+            3 => Some(Name::ObjectAccessedGetLegalHold),
+            4 => Some(Name::ObjectAccessedHead),
+            5 => Some(Name::ObjectAccessedAttributes),
+            6 => Some(Name::ObjectCreatedCompleteMultipartUpload),
+            7 => Some(Name::ObjectCreatedCopy),
+            8 => Some(Name::ObjectCreatedPost),
+            9 => Some(Name::ObjectCreatedPut),
+            10 => Some(Name::ObjectCreatedPutRetention),
+            11 => Some(Name::ObjectCreatedPutLegalHold),
+            12 => Some(Name::ObjectCreatedPutTagging),
+            13 => Some(Name::ObjectCreatedDeleteTagging),
+            14 => Some(Name::ObjectRemovedDelete),
+            15 => Some(Name::ObjectRemovedDeleteMarkerCreated),
+            16 => Some(Name::ObjectRemovedDeleteAllVersions),
+            17 => Some(Name::ObjectRemovedNoOp),
+            18 => Some(Name::BucketCreated),
+            19 => Some(Name::BucketRemoved),
+            20 => Some(Name::ObjectReplicationFailed),
+            21 => Some(Name::ObjectReplicationComplete),
+            22 => Some(Name::ObjectReplicationMissedThreshold),
+            23 => Some(Name::ObjectReplicationReplicatedAfterThreshold),
+            24 => Some(Name::ObjectReplicationNotTracked),
+            25 => Some(Name::ObjectRestorePost),
+            26 => Some(Name::ObjectRestoreCompleted),
+            27 => Some(Name::ObjectTransitionFailed),
+            28 => Some(Name::ObjectTransitionComplete),
+            29 => Some(Name::ObjectManyVersions),
+            30 => Some(Name::ObjectLargeVersions),
+            31 => Some(Name::PrefixManyFolders),
+            32 => Some(Name::IlmDelMarkerExpirationDelete),
+            33 => Some(Name::ObjectAccessedAll),
+            34 => Some(Name::ObjectCreatedAll),
+            35 => Some(Name::ObjectRemovedAll),
+            36 => Some(Name::ObjectReplicationAll),
+            37 => Some(Name::ObjectRestoreAll),
+            38 => Some(Name::ObjectTransitionAll),
+            39 => Some(Name::ObjectScannerAll),
+            40 => Some(Name::Everything),
+            _ => None,
+        }
+    }
+}

crates/event/src/lib.rs

@@ -0,0 +1,4 @@
+mod error;
+mod event;
+mod notifier;
+mod system;

crates/event/src/notifier.rs

@@ -0,0 +1,143 @@
+use crate::config::EventNotifierConfig;
+use crate::event::Event;
+use common::error::{Error, Result};
+use ecstore::store::ECStore;
+use std::sync::Arc;
+use tokio::sync::{broadcast, mpsc};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, instrument, warn};
+
+/// Event Notifier
+pub struct EventNotifier {
+    /// The event sending channel
+    sender: mpsc::Sender<Event>,
+    /// Receiver task handle
+    task_handle: Option<tokio::task::JoinHandle<()>>,
+    /// Configuration information
+    config: EventNotifierConfig,
+    /// Turn off tagging
+    shutdown: CancellationToken,
+    /// Close the notification channel
+    shutdown_complete_tx: Option<broadcast::Sender<()>>,
+}
+
+impl EventNotifier {
+    /// Create a new event notifier
+    #[instrument(skip_all)]
+    pub async fn new(store: Arc<ECStore>) -> Result<Self> {
+        let manager = crate::store::manager::EventManager::new(store);
+
+        let manager = Arc::new(manager.await);
+
+        // Initialize the configuration
+        let config = manager.clone().init().await?;
+
+        // Create adapters
+        let adapters = manager.clone().create_adapters().await?;
+        info!("Created {} adapters", adapters.len());
+
+        // Create a close marker
+        let shutdown = CancellationToken::new();
+        let (shutdown_complete_tx, _) = broadcast::channel(1);
+
+        // 创建事件通道 - 使用默认容量，因为每个适配器都有自己的队列
+        // 这里使用较小的通道容量，因为事件会被快速分发到适配器
+        let (sender, mut receiver) = mpsc::channel::<Event>(100);
+
+        let shutdown_clone = shutdown.clone();
+        let shutdown_complete_tx_clone = shutdown_complete_tx.clone();
+        let adapters_clone = adapters.clone();
+
+        // Start the event processing task
+        let task_handle = tokio::spawn(async move {
+            debug!("The event processing task starts");
+
+            loop {
+                tokio::select! {
+                    Some(event) = receiver.recv() => {
+                        debug!("The event is received:{}", event.id);
+
+                        // Distribute to all adapters
+                        for adapter in &adapters_clone {
+                            let adapter_name = adapter.name();
+                            match adapter.send(&event).await {
+                                Ok(_) => {
+                                    debug!("Event {} Successfully sent to the adapter {}", event.id, adapter_name);
+                                }
+                                Err(e) => {
+                                    error!("Event {} send to adapter {} failed:{}", event.id, adapter_name, e);
+                                }
+                            }
+                        }
+                    }
+
+                    _ = shutdown_clone.cancelled() => {
+                        info!("A shutdown signal is received, and the event processing task is stopped");
+                        let _ = shutdown_complete_tx_clone.send(());
+                        break;
+                    }
+                }
+            }
+
+            debug!("The event processing task has been stopped");
+        });
+
+        Ok(Self {
+            sender,
+            task_handle: Some(task_handle),
+            config,
+            shutdown,
+            shutdown_complete_tx: Some(shutdown_complete_tx),
+        })
+    }
+
+    /// Turn off the event notifier
+    pub async fn shutdown(&mut self) -> Result<()> {
+        info!("Turn off the event notifier");
+        self.shutdown.cancel();
+
+        if let Some(shutdown_tx) = self.shutdown_complete_tx.take() {
+            let mut rx = shutdown_tx.subscribe();
+
+            // Wait for the shutdown to complete the signal or time out
+            tokio::select! {
+                _ = rx.recv() => {
+                    debug!("A shutdown completion signal is received");
+                }
+                _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
+                    warn!("Shutdown timeout and forced termination");
+                }
+            }
+        }
+
+        if let Some(handle) = self.task_handle.take() {
+            handle.abort();
+            match handle.await {
+                Ok(_) => debug!("The event processing task has been terminated gracefully"),
+                Err(e) => {
+                    if e.is_cancelled() {
+                        debug!("The event processing task has been canceled");
+                    } else {
+                        error!("An error occurred while waiting for the event processing task to terminate:{}", e);
+                    }
+                }
+            }
+        }
+
+        info!("The event notifier is completely turned off");
+        Ok(())
+    }
+
+    /// Send events
+    pub async fn send(&self, event: Event) -> Result<()> {
+        self.sender
+            .send(event)
+            .await
+            .map_err(|e| Error::msg(format!("Failed to send events to channel:{}", e)))
+    }
+
+    /// Get the current configuration
+    pub fn config(&self) -> &EventNotifierConfig {
+        &self.config
+    }
+}

crates/event/src/system.rs

@@ -0,0 +1,80 @@
+use crate::notifier::EventNotifier;
+use common::error::Result;
+use ecstore::store::ECStore;
+use once_cell::sync::OnceCell;
+use std::sync::{Arc, Mutex};
+use tracing::{debug, error, info};
+
+/// Global event system
+pub struct EventSystem {
+    /// Event Notifier
+    notifier: Mutex<Option<EventNotifier>>,
+}
+
+impl EventSystem {
+    /// Create a new event system
+    pub fn new() -> Self {
+        Self {
+            notifier: Mutex::new(None),
+        }
+    }
+
+    /// Initialize the event system
+    pub async fn init(&self, store: Arc<ECStore>) -> Result<EventNotifierConfig> {
+        info!("Initialize the event system");
+        let notifier = EventNotifier::new(store).await?;
+        let config = notifier.config().clone();
+
+        let mut guard = self
+            .notifier
+            .lock()
+            .map_err(|e| common::error::Error::msg(format!("Failed to acquire locks:{}", e)))?;
+
+        *guard = Some(notifier);
+        debug!("The event system initialization is complete");
+
+        Ok(config)
+    }
+
+    /// Send events
+    pub async fn send_event(&self, event: crate::Event) -> Result<()> {
+        let guard = self
+            .notifier
+            .lock()
+            .map_err(|e| common::error::Error::msg(format!("Failed to acquire locks:{}", e)))?;
+
+        if let Some(notifier) = &*guard {
+            notifier.send(event).await
+        } else {
+            error!("The event system is not initialized");
+            Err(common::error::Error::msg("The event system is not initialized"))
+        }
+    }
+
+    /// Shut down the event system
+    pub async fn shutdown(&self) -> Result<()> {
+        info!("Shut down the event system");
+        let mut guard = self
+            .notifier
+            .lock()
+            .map_err(|e| common::error::Error::msg(format!("Failed to acquire locks:{}", e)))?;
+
+        if let Some(ref mut notifier) = *guard {
+            notifier.shutdown().await?;
+            *guard = None;
+            info!("The event system is down");
+            Ok(())
+        } else {
+            debug!("The event system has been shut down");
+            Ok(())
+        }
+    }
+}
+
+/// A global event system instance
+pub static GLOBAL_EVENT_SYS: OnceCell<EventSystem> = OnceCell::new();
+
+/// Initialize the global event system
+pub fn init_global_event_system() -> &'static EventSystem {
+    GLOBAL_EVENT_SYS.get_or_init(EventSystem::new)
+}

crates/notify/Cargo.toml

@@ -10,13 +10,12 @@ version.workspace = true
 default = ["webhook"]
 webhook = ["dep:reqwest"]
 mqtt = ["rumqttc"]
-kafka = ["dep:rdkafka"]
 
 [dependencies]
+rustfs-config = { workspace = true, features = ["constants", "notify"] }
 async-trait = { workspace = true }
 common = { workspace = true }
 ecstore = { workspace = true }
-lazy_static = { workspace = true }
 once_cell = { workspace = true }
 reqwest = { workspace = true, optional = true }
 rumqttc = { workspace = true, optional = true }
@@ -32,11 +31,6 @@ tokio-util = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde"] }
 snap = { workspace = true }
 
-
-# Only enable kafka features and related dependencies on Linux
-[target.'cfg(target_os = "linux")'.dependencies]
-rdkafka = { workspace = true, features = ["tokio"], optional = true }
-
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }
 tracing-subscriber = { workspace = true }

crates/notify/src/adapter/kafka.rs

@@ -1,180 +0,0 @@
-use crate::config::kafka::KafkaConfig;
-use crate::config::{default_queue_limit, DEFAULT_RETRY_INTERVAL, STORE_PREFIX};
-use crate::{ChannelAdapter, ChannelAdapterType};
-use crate::{Error, Event, QueueStore};
-use async_trait::async_trait;
-use rdkafka::error::KafkaError;
-use rdkafka::producer::{FutureProducer, FutureRecord};
-use rdkafka::types::RDKafkaErrorCode;
-use rdkafka::util::Timeout;
-use std::path::PathBuf;
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::time::sleep;
-use ChannelAdapterType::Kafka;
-
-/// Kafka adapter for sending events to a Kafka topic.
-pub struct KafkaAdapter {
-    producer: FutureProducer,
-    store: Option<Arc<QueueStore<Event>>>,
-    config: KafkaConfig,
-}
-
-impl KafkaAdapter {
-    /// Creates a new Kafka adapter.
-    pub fn new(config: &KafkaConfig) -> Result<Self, Error> {
-        // Create a Kafka producer with the provided configuration.
-        let producer = rdkafka::config::ClientConfig::new()
-            .set("bootstrap.servers", &config.brokers)
-            .set("message.timeout.ms", config.timeout.to_string())
-            .create()
-            .map_err(|e| Error::msg(format!("Failed to create a Kafka producer: {}", e)))?;
-
-        // create a queue store if enabled
-        let store = if !config.common.queue_dir.is_empty() {
-            let store_path = PathBuf::from(&config.common.queue_dir).join(format!(
-                "{}-{}-{}",
-                STORE_PREFIX,
-                Kafka.as_str(),
-                config.common.identifier
-            ));
-
-            let queue_limit = if config.queue_limit > 0 {
-                config.queue_limit
-            } else {
-                default_queue_limit()
-            };
-            let store = QueueStore::new(store_path, config.queue_limit, Some(".event".to_string()));
-            if let Err(e) = store.open() {
-                tracing::error!("Unable to open queue storage: {}", e);
-                None
-            } else {
-                Some(Arc::new(store))
-            }
-        } else {
-            None
-        };
-
-        Ok(Self { config, producer, store })
-    }
-
-    /// Handle backlog events in storage
-    pub async fn process_backlog(&self) -> Result<(), Error> {
-        if let Some(store) = &self.store {
-            let keys = store.list();
-
-            for key in keys {
-                match store.get_multiple(&key) {
-                    Ok(events) => {
-                        for event in events {
-                            // Use the retry interval to send events
-                            if let Err(e) = self.send_with_retry(&event).await {
-                                tracing::error!("Processing of backlog events failed: {}", e);
-                                // If it still fails, we remain in the queue
-                                break;
-                            }
-                        }
-
-                        // The event is deleted after it has been successfully processed
-                        if let Err(e) = store.del(&key) {
-                            tracing::error!("Failed to delete a handled event: {}", e);
-                        }
-                    }
-                    Err(e) => {
-                        tracing::error!("Fetch events from the queue failed: {}", e);
-
-                        // If the event cannot be read, it may be corrupted, delete it
-                        if let Err(del_err) = store.del(&key) {
-                            tracing::error!("Failed to delete a corrupted event: {}", del_err);
-                        }
-                    }
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    /// Sends an event to the Kafka topic with retry logic.
-    async fn send_with_retry(&self, event: &Event) -> Result<(), Error> {
-        let retry_interval = match self.config.retry_interval {
-            Some(t) => Duration::from_secs(t),
-            None => Duration::from_secs(DEFAULT_RETRY_INTERVAL), // Default to 3 seconds if not set
-        };
-
-        for attempt in 0..self.max_retries {
-            match self.send_request(event).await {
-                Ok(_) => return Ok(()),
-                Err((KafkaError::MessageProduction(RDKafkaErrorCode::QueueFull), _)) => {
-                    tracing::warn!("Kafka attempt {} failed: Queue full. Retrying...", attempt + 1);
-                    // sleep(Duration::from_secs(2u64.pow(attempt))).await;
-                    sleep(retry_interval).await;
-                }
-                Err((e, _)) => {
-                    tracing::error!("Kafka send error: {}", e);
-                    return Err(Error::Kafka(e));
-                }
-            }
-        }
-
-        Err(Error::Custom("Exceeded maximum retry attempts for Kafka message".to_string()))
-    }
-
-    /// Send a single Kafka message
-    async fn send_request(&self, event: &Event) -> Result<(), Error> {
-        // Serialize events
-        let payload = serde_json::to_string(event).map_err(|e| Error::Custom(format!("Serialization event failed: {}", e)))?;
-
-        // Create a Kafka record
-        let record = FutureRecord::to(&self.config.topic).payload(&payload).key(&event.id); // Use the event ID as the key
-
-        // Send to Kafka
-        let delivery_status = self
-            .producer
-            .send(record, Duration::from_millis(self.config.timeout))
-            .await
-            .map_err(|(e, _)| Error::Custom(format!("Failed to send to Kafka: {}", e)))?;
-        // Check delivery status
-        if let Some((err, _)) = delivery_status {
-            return Err(Error::Kafka(err));
-        }
-
-        Ok(())
-    }
-
-    /// Save the event to the queue
-    async fn save_to_queue(&self, event: &Event) -> Result<(), Error> {
-        if let Some(store) = &self.store {
-            store
-                .put(event.clone())
-                .map_err(|e| Error::Custom(format!("Saving events to queue failed: {}", e)))?;
-        }
-        Ok(())
-    }
-}
-
-#[async_trait]
-impl ChannelAdapter for KafkaAdapter {
-    fn name(&self) -> String {
-        ChannelAdapterType::Kafka.to_string()
-    }
-
-    async fn send(&self, event: &Event) -> Result<(), Error> {
-        // Try to deal with the backlog of events first
-        let _ = self.process_backlog().await;
-
-        // An attempt was made to send the current event
-        match self.send_with_retry(event).await {
-            Ok(_) => Ok(()),
-            Err(e) => {
-                // If the send fails and the queue is enabled, save to the queue
-                if let Some(_) = &self.store {
-                    tracing::warn!("Failed to send events to Kafka and saved to a queue: {}", e);
-                    self.save_to_queue(event).await?;
-                    return Ok(());
-                }
-                Err(e)
-            }
-        }
-    }
-}

crates/notify/src/adapter/mod.rs

@@ -1,10 +1,8 @@
-use crate::config::adapter::AdapterConfig;
+use crate::config::AdapterConfig;
 use crate::{Error, Event};
 use async_trait::async_trait;
 use std::sync::Arc;
 
-#[cfg(all(feature = "kafka", target_os = "linux"))]
-pub(crate) mod kafka;
 #[cfg(feature = "mqtt")]
 pub(crate) mod mqtt;
 #[cfg(feature = "webhook")]
@@ -97,10 +95,6 @@ pub fn create_adapters(configs: Vec<AdapterConfig>) -> Result<Vec<Arc<dyn Channe
                 webhook_config.validate().map_err(Error::ConfigError)?;
                 adapters.push(Arc::new(webhook::WebhookAdapter::new(webhook_config.clone())));
             }
-            #[cfg(all(feature = "kafka", target_os = "linux"))]
-            AdapterConfig::Kafka(kafka_config) => {
-                adapters.push(Arc::new(kafka::KafkaAdapter::new(kafka_config)?));
-            }
             #[cfg(feature = "mqtt")]
             AdapterConfig::Mqtt(mqtt_config) => {
                 let (mqtt, mut event_loop) = mqtt::MqttAdapter::new(mqtt_config);
@@ -109,8 +103,6 @@ pub fn create_adapters(configs: Vec<AdapterConfig>) -> Result<Vec<Arc<dyn Channe
             }
             #[cfg(not(feature = "webhook"))]
             AdapterConfig::Webhook(_) => return Err(Error::FeatureDisabled("webhook")),
-            #[cfg(any(not(feature = "kafka"), not(target_os = "linux")))]
-            AdapterConfig::Kafka(_) => return Err(Error::FeatureDisabled("kafka")),
             #[cfg(not(feature = "mqtt"))]
             AdapterConfig::Mqtt(_) => return Err(Error::FeatureDisabled("mqtt")),
         }

crates/notify/src/adapter/webhook.rs

@@ -1,8 +1,5 @@
-use crate::config::webhook::WebhookConfig;
 use crate::config::STORE_PREFIX;
-use crate::store::queue::Store;
 use crate::{ChannelAdapter, ChannelAdapterType};
-use crate::{Error, QueueStore};
 use crate::{Event, DEFAULT_RETRY_INTERVAL};
 use async_trait::async_trait;
 use reqwest::header::{HeaderMap, HeaderName, HeaderValue};

crates/notify/src/config.rs

@@ -1,20 +1,59 @@
-use crate::config::{adapter::AdapterConfig, kafka::KafkaConfig, mqtt::MqttConfig, webhook::WebhookConfig};
+use rustfs_config::notify::mqtt::MQTTArgs;
+use rustfs_config::notify::webhook::WebhookArgs;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
+use std::env;
+
+/// The default configuration file name
+const DEFAULT_CONFIG_FILE: &str = "notify";
+
+/// The prefix for the configuration file
+pub const STORE_PREFIX: &str = "rustfs";
+
+/// The default retry interval for the webhook adapter
+pub const DEFAULT_RETRY_INTERVAL: u64 = 3;
+
+/// The default maximum retry count for the webhook adapter
+pub const DEFAULT_MAX_RETRIES: u32 = 3;
+
+/// The default notification queue limit
+pub const DEFAULT_NOTIFY_QUEUE_LIMIT: u64 = 10000;
+
+/// Provide temporary directories as default storage paths
+pub(crate) fn default_queue_dir() -> String {
+    env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| {
+        tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string());
+        env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string()
+    })
+}
+
+/// Provides the recommended default channel capacity for high concurrency systems
+pub(crate) fn default_queue_limit() -> u64 {
+    env::var("EVENT_CHANNEL_CAPACITY")
+        .unwrap_or_else(|_| DEFAULT_NOTIFY_QUEUE_LIMIT.to_string())
+        .parse()
+        .unwrap_or(DEFAULT_NOTIFY_QUEUE_LIMIT) // Default to 10000 if parsing fails
+}
+
+/// Configuration for the adapter.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "type")]
+pub enum AdapterConfig {
+    Webhook(WebhookArgs),
+    Mqtt(MQTTArgs),
+}
+
 /// Event Notifier Configuration
 /// This struct contains the configuration for the event notifier system,
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct EventNotifierConfig {
     /// A collection of webhook configurations, with the key being a unique identifier
     #[serde(default)]
-    pub webhook: HashMap<String, WebhookConfig>,
-    /// A collection of Kafka configurations, with the key being a unique identifier
-    #[serde(default)]
-    pub kafka: HashMap<String, KafkaConfig>,
+    pub webhook: HashMap<String, WebhookArgs>,
     ///MQTT configuration collection, with the key being a unique identifier
     #[serde(default)]
-    pub mqtt: HashMap<String, MqttConfig>,
+    pub mqtt: HashMap<String, MQTTArgs>,
 }
 
 impl EventNotifierConfig {
@@ -49,21 +88,14 @@ impl EventNotifierConfig {
 
         // Add all enabled webhook configurations
         for webhook in self.webhook.values() {
-            if webhook.common.enable {
+            if webhook.enable {
                 adapters.push(AdapterConfig::Webhook(webhook.clone()));
             }
         }
 
-        // Add all enabled Kafka configurations
-        for kafka in self.kafka.values() {
-            if kafka.common.enable {
-                adapters.push(AdapterConfig::Kafka(kafka.clone()));
-            }
-        }
-
         // Add all enabled MQTT configurations
         for mqtt in self.mqtt.values() {
-            if mqtt.common.enable {
+            if mqtt.enable {
                 adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
             }
         }

crates/notify/src/config/adapter.rs

@@ -1,42 +0,0 @@
-use crate::config::kafka::KafkaConfig;
-use crate::config::mqtt::MqttConfig;
-use crate::config::webhook::WebhookConfig;
-use crate::config::{default_queue_dir, default_queue_limit};
-use serde::{Deserialize, Serialize};
-
-/// Add a common field for the adapter configuration
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct AdapterCommon {
-    /// Adapter identifier for unique identification
-    pub identifier: String,
-    /// Adapter description information
-    pub comment: String,
-    /// Whether to enable this adapter
-    #[serde(default)]
-    pub enable: bool,
-    #[serde(default = "default_queue_dir")]
-    pub queue_dir: String,
-    #[serde(default = "default_queue_limit")]
-    pub queue_limit: u64,
-}
-
-impl Default for AdapterCommon {
-    fn default() -> Self {
-        Self {
-            identifier: String::new(),
-            comment: String::new(),
-            enable: false,
-            queue_dir: default_queue_dir(),
-            queue_limit: default_queue_limit(),
-        }
-    }
-}
-
-/// Configuration for the adapter.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-#[serde(tag = "type")]
-pub enum AdapterConfig {
-    Webhook(WebhookConfig),
-    Kafka(KafkaConfig),
-    Mqtt(MqttConfig),
-}

crates/notify/src/config/kafka.rs

@@ -1,44 +0,0 @@
-use crate::config::adapter::AdapterCommon;
-use crate::config::{default_queue_dir, default_queue_limit};
-use serde::{Deserialize, Serialize};
-
-/// Configuration for the Kafka adapter.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct KafkaConfig {
-    #[serde(flatten)]
-    pub common: AdapterCommon,
-    pub brokers: String,
-    pub topic: String,
-    pub max_retries: u32,
-    pub timeout: u64,
-}
-
-impl Default for KafkaConfig {
-    fn default() -> Self {
-        Self {
-            common: AdapterCommon::default(),
-            brokers: String::new(),
-            topic: String::new(),
-            max_retries: 3,
-            timeout: 5000,
-        }
-    }
-}
-
-impl KafkaConfig {
-    /// Create a new Kafka configuration
-    pub fn new(identifier: impl Into<String>, brokers: impl Into<String>, topic: impl Into<String>) -> Self {
-        Self {
-            common: AdapterCommon {
-                identifier: identifier.into(),
-                comment: String::new(),
-                enable: true,
-                queue_dir: default_queue_dir(),
-                queue_limit: default_queue_limit(),
-            },
-            brokers: brokers.into(),
-            topic: topic.into(),
-            ..Default::default()
-        }
-    }
-}

crates/notify/src/config/mod.rs

@@ -1,38 +0,0 @@
-use std::env;
-
-pub mod adapter;
-pub mod kafka;
-pub mod mqtt;
-pub mod notifier;
-pub mod webhook;
-
-/// The default configuration file name
-const DEFAULT_CONFIG_FILE: &str = "event";
-
-/// The prefix for the configuration file
-pub const STORE_PREFIX: &str = "rustfs";
-
-/// The default retry interval for the webhook adapter
-pub const DEFAULT_RETRY_INTERVAL: u64 = 3;
-
-/// The default maximum retry count for the webhook adapter
-pub const DEFAULT_MAX_RETRIES: u32 = 3;
-
-/// The default notification queue limit
-pub const DEFAULT_NOTIFY_QUEUE_LIMIT: u64 = 10000;
-
-/// Provide temporary directories as default storage paths
-pub(crate) fn default_queue_dir() -> String {
-    env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| {
-        tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string());
-        env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string()
-    })
-}
-
-/// Provides the recommended default channel capacity for high concurrency systems
-pub(crate) fn default_queue_limit() -> u64 {
-    env::var("EVENT_CHANNEL_CAPACITY")
-        .unwrap_or_else(|_| DEFAULT_NOTIFY_QUEUE_LIMIT.to_string())
-        .parse()
-        .unwrap_or(DEFAULT_NOTIFY_QUEUE_LIMIT) // Default to 10000 if parsing fails
-}

crates/notify/src/config/mqtt.rs

@@ -1,46 +0,0 @@
-use crate::config::adapter::AdapterCommon;
-use crate::config::{default_queue_dir, default_queue_limit};
-use serde::{Deserialize, Serialize};
-
-/// Configuration for the MQTT adapter.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct MqttConfig {
-    #[serde(flatten)]
-    pub common: AdapterCommon,
-    pub broker: String,
-    pub port: u16,
-    pub client_id: String,
-    pub topic: String,
-    pub max_retries: u32,
-}
-
-impl Default for MqttConfig {
-    fn default() -> Self {
-        Self {
-            common: AdapterCommon::default(),
-            broker: String::new(),
-            port: 1883,
-            client_id: String::new(),
-            topic: String::new(),
-            max_retries: 3,
-        }
-    }
-}
-
-impl MqttConfig {
-    /// Create a new MQTT configuration
-    pub fn new(identifier: impl Into<String>, broker: impl Into<String>, topic: impl Into<String>) -> Self {
-        Self {
-            common: AdapterCommon {
-                identifier: identifier.into(),
-                comment: String::new(),
-                enable: true,
-                queue_dir: default_queue_dir(),
-                queue_limit: default_queue_limit(),
-            },
-            broker: broker.into(),
-            topic: topic.into(),
-            ..Default::default()
-        }
-    }
-}

crates/notify/src/config/webhook.rs

@@ -1,88 +0,0 @@
-use crate::config::adapter::AdapterCommon;
-use crate::config::{default_queue_dir, default_queue_limit};
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::path::Path;
-use tracing::info;
-
-/// Configuration for the webhook adapter.
-#[derive(Debug, Clone, Serialize, Deserialize, Default)]
-pub struct WebhookConfig {
-    #[serde(flatten)]
-    pub common: AdapterCommon,
-    pub endpoint: String,
-    pub auth_token: Option<String>,
-    pub custom_headers: Option<HashMap<String, String>>,
-    pub max_retries: u32,
-    pub retry_interval: Option<u64>,
-    pub timeout: Option<u64>,
-    #[serde(default)]
-    pub client_cert: Option<String>,
-    #[serde(default)]
-    pub client_key: Option<String>,
-}
-
-impl WebhookConfig {
-    /// validate the configuration for the webhook adapter
-    ///
-    /// # Returns
-    ///
-    /// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid.
-    ///
-    /// # Errors
-    /// - Returns an error if the configuration is invalid, such as empty endpoint, unreasonable timeout, or mismatched certificate and key.
-    pub fn validate(&self) -> Result<(), String> {
-        // If not enabled, the other fields are not validated
-        if !self.common.enable {
-            return Ok(());
-        }
-
-        // verify that endpoint cannot be empty
-        if self.endpoint.trim().is_empty() {
-            return Err("Webhook endpoint cannot be empty".to_string());
-        }
-
-        // verification timeout must be reasonable
-        if self.timeout.is_some() {
-            match self.timeout {
-                Some(timeout) if timeout > 0 => {
-                    info!("Webhook timeout is set to {}", timeout);
-                }
-                _ => return Err("Webhook timeout must be greater than 0".to_string()),
-            }
-        }
-
-        // Verify that the maximum number of retry is reasonable
-        if self.max_retries > 10 {
-            return Err("Maximum retry count cannot exceed 10".to_string());
-        }
-
-        // Verify the queue directory path
-        if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() {
-            return Err("Queue directory path should be absolute".to_string());
-        }
-
-        // The authentication certificate and key must appear in pairs
-        if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some())
-        {
-            return Err("Certificate and key must be specified as a pair".to_string());
-        }
-
-        Ok(())
-    }
-
-    /// Create a new webhook configuration
-    pub fn new(identifier: impl Into<String>, endpoint: impl Into<String>) -> Self {
-        Self {
-            common: AdapterCommon {
-                identifier: identifier.into(),
-                comment: String::new(),
-                enable: true,
-                queue_dir: default_queue_dir(),
-                queue_limit: default_queue_limit(),
-            },
-            endpoint: endpoint.into(),
-            ..Default::default()
-        }
-    }
-}

crates/notify/src/lib.rs

@@ -7,8 +7,6 @@ mod store;
 mod system;
 
 pub use adapter::create_adapters;
-#[cfg(all(feature = "kafka", target_os = "linux"))]
-pub use adapter::kafka::KafkaAdapter;
 #[cfg(feature = "mqtt")]
 pub use adapter::mqtt::MqttAdapter;
 #[cfg(feature = "webhook")]
@@ -16,18 +14,6 @@ pub use adapter::webhook::WebhookAdapter;
 
 pub use adapter::ChannelAdapter;
 pub use adapter::ChannelAdapterType;
-pub use config::adapter::AdapterCommon;
-pub use config::adapter::AdapterConfig;
-pub use config::notifier::EventNotifierConfig;
-pub use config::{DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL};
+pub use config::{AdapterConfig, EventNotifierConfig, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL};
 pub use error::Error;
 pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source};
-pub use store::queue::QueueStore;
-
-#[cfg(all(feature = "kafka", target_os = "linux"))]
-pub use config::kafka::KafkaConfig;
-#[cfg(feature = "mqtt")]
-pub use config::mqtt::MqttConfig;
-
-#[cfg(feature = "webhook")]
-pub use config::webhook::WebhookConfig;

crates/notify/src/notifier.rs

@@ -1,4 +1,4 @@
-use crate::config::notifier::EventNotifierConfig;
+use crate::config::EventNotifierConfig;
 use crate::Event;
 use common::error::{Error, Result};
 use ecstore::store::ECStore;

crates/notify/src/store/manager.rs

@@ -115,11 +115,6 @@ impl EventManager {
             merged.webhook.insert(id, config);
         }
 
-        // Merge Kafka configurations
-        for (id, config) in new.kafka {
-            merged.kafka.insert(id, config);
-        }
-
         // Merge MQTT configurations
         for (id, config) in new.mqtt {
             merged.mqtt.insert(id, config);

crates/notify/src/store/mod.rs

@@ -1,2 +1,314 @@
+use async_trait::async_trait;
+use serde::{de::DeserializeOwned, Serialize};
+use std::error::Error;
+use std::fmt;
+use std::fmt::Display;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tokio::time;
+
 pub(crate) mod manager;
 pub(crate) mod queue;
+
+// 常量定义
+pub const RETRY_INTERVAL: Duration = Duration::from_secs(3);
+pub const DEFAULT_LIMIT: u64 = 100000; // 默认存储限制
+pub const DEFAULT_EXT: &str = ".unknown";
+pub const COMPRESS_EXT: &str = ".snappy";
+
+// 错误类型
+#[derive(Debug)]
+pub enum StoreError {
+    NotConnected,
+    LimitExceeded,
+    IoError(std::io::Error),
+    Utf8(std::str::Utf8Error),
+    SerdeError(serde_json::Error),
+    Deserialize(serde_json::Error),
+    UuidError(uuid::Error),
+    Other(String),
+}
+
+impl Display for StoreError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            StoreError::NotConnected => write!(f, "not connected to target server/service"),
+            StoreError::LimitExceeded => write!(f, "the maximum store limit reached"),
+            StoreError::IoError(e) => write!(f, "IO error: {}", e),
+            StoreError::Utf8(e) => write!(f, "UTF-8 conversion error: {}", e),
+            StoreError::SerdeError(e) => write!(f, "serialization error: {}", e),
+            StoreError::Deserialize(e) => write!(f, "deserialization error: {}", e),
+            StoreError::UuidError(e) => write!(f, "UUID generation error: {}", e),
+            StoreError::Other(s) => write!(f, "{}", s),
+        }
+    }
+}
+
+impl Error for StoreError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            StoreError::IoError(e) => Some(e),
+            StoreError::SerdeError(e) => Some(e),
+            StoreError::UuidError(e) => Some(e),
+            _ => None,
+        }
+    }
+}
+
+impl From<std::io::Error> for StoreError {
+    fn from(e: std::io::Error) -> Self {
+        StoreError::IoError(e)
+    }
+}
+
+impl From<serde_json::Error> for StoreError {
+    fn from(e: serde_json::Error) -> Self {
+        StoreError::SerdeError(e)
+    }
+}
+
+impl From<uuid::Error> for StoreError {
+    fn from(e: uuid::Error) -> Self {
+        StoreError::UuidError(e)
+    }
+}
+
+pub type StoreResult<T> = Result<T, StoreError>;
+
+// 日志记录器类型
+pub type Logger = fn(ctx: Option<&str>, err: StoreError, id: &str, err_kind: &[&dyn Display]);
+
+// Key 结构体定义
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Key {
+    pub name: String,
+    pub compress: bool,
+    pub extension: String,
+    pub item_count: usize,
+}
+
+impl Key {
+    pub fn new(name: String, extension: String) -> Self {
+        Self {
+            name,
+            extension,
+            compress: false,
+            item_count: 1,
+        }
+    }
+
+    pub fn with_compression(mut self, compress: bool) -> Self {
+        self.compress = compress;
+        self
+    }
+
+    pub fn with_item_count(mut self, count: usize) -> Self {
+        self.item_count = count;
+        self
+    }
+
+    pub fn to_string(&self) -> String {
+        let mut key_str = self.name.clone();
+
+        if self.item_count > 1 {
+            key_str = format!("{}:{}", self.item_count, self.name);
+        }
+
+        let ext = if self.compress {
+            format!("{}{}", self.extension, COMPRESS_EXT)
+        } else {
+            self.extension.clone()
+        };
+
+        format!("{}{}", key_str, ext)
+    }
+}
+
+impl Display for Key {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.to_string())
+    }
+}
+
+pub fn parse_key(k: &str) -> Key {
+    let mut key = Key {
+        name: k.to_string(),
+        compress: false,
+        extension: String::new(),
+        item_count: 1,
+    };
+
+    // 检查压缩扩展名
+    if k.ends_with(COMPRESS_EXT) {
+        key.compress = true;
+        key.name = key.name[..key.name.len() - COMPRESS_EXT.len()].to_string();
+    }
+
+    // 解析项目数量
+    if let Some(colon_pos) = key.name.find(':') {
+        if let Ok(count) = key.name[..colon_pos].parse::<usize>() {
+            key.item_count = count;
+            key.name = key.name[colon_pos + 1..].to_string();
+        }
+    }
+
+    // 解析扩展名
+    if let Some(dot_pos) = key.name.rfind('.') {
+        key.extension = key.name[dot_pos..].to_string();
+        key.name = key.name[..dot_pos].to_string();
+    }
+
+    key
+}
+
+// Target trait 定义
+#[async_trait]
+pub trait Target: Send + Sync {
+    fn name(&self) -> String;
+    async fn send_from_store(&self, key: Key) -> StoreResult<()>;
+}
+
+// Store trait 定义
+#[async_trait]
+pub trait Store<T>: Send + Sync
+where
+    T: Serialize + DeserializeOwned + Send + Sync + 'static,
+{
+    async fn put(&self, item: T) -> StoreResult<Key>;
+    async fn put_multiple(&self, items: Vec<T>) -> StoreResult<Key>;
+    async fn get(&self, key: Key) -> StoreResult<T>;
+    async fn get_multiple(&self, key: Key) -> StoreResult<Vec<T>>;
+    async fn get_raw(&self, key: Key) -> StoreResult<Vec<u8>>;
+    async fn put_raw(&self, b: Vec<u8>) -> StoreResult<Key>;
+    async fn len(&self) -> usize;
+    async fn list(&self) -> Vec<Key>;
+    async fn del(&self, key: Key) -> StoreResult<()>;
+    async fn open(&self) -> StoreResult<()>;
+    async fn delete(&self) -> StoreResult<()>;
+}
+
+// 重播项目辅助函数
+pub async fn replay_items<T>(store: Arc<dyn Store<T>>, done_ch: mpsc::Receiver<()>, log: Logger, id: &str) -> mpsc::Receiver<Key>
+where
+    T: Serialize + DeserializeOwned + Send + Sync + 'static,
+{
+    let (tx, rx) = mpsc::channel(100); // 合理的缓冲区大小
+    let id = id.to_string();
+
+    tokio::spawn(async move {
+        let mut done_ch = done_ch;
+        let mut retry_interval = time::interval(RETRY_INTERVAL);
+        let mut retry_interval = time::interval_at(retry_interval.tick().await, RETRY_INTERVAL);
+
+        loop {
+            let keys = store.list().await;
+
+            for key in keys {
+                let tx = tx.clone();
+                tokio::select! {
+                    _ = tx.send(key) => {
+                        // 成功发送下一个键
+                    }
+                    _ = done_ch.recv() => {
+                        return;
+                    }
+                }
+            }
+
+            tokio::select! {
+                _ = retry_interval.tick() => {
+                    // 重试定时器触发，继续循环
+                }
+                _ = done_ch.recv() => {
+                    return;
+                }
+            }
+        }
+    });
+
+    rx
+}
+
+// 发送项目辅助函数
+pub async fn send_items(target: &dyn Target, mut key_ch: mpsc::Receiver<Key>, mut done_ch: mpsc::Receiver<()>, logger: Logger) {
+    let mut retry_interval = time::interval(RETRY_INTERVAL);
+
+    async fn try_send(
+        target: &dyn Target,
+        key: Key,
+        retry_interval: &mut time::Interval,
+        done_ch: &mut mpsc::Receiver<()>,
+        logger: Logger,
+    ) -> bool {
+        loop {
+            match target.send_from_store(key.clone()).await {
+                Ok(_) => return true,
+                Err(err) => {
+                    logger(None, err, &target.name(), &[&format!("unable to send log entry to '{}'", target.name())]);
+
+                    tokio::select! {
+                        _ = retry_interval.tick() => {
+                            // 重试
+                        }
+                        _ = done_ch.recv() => {
+                            return false;
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    loop {
+        tokio::select! {
+            maybe_key = key_ch.recv() => {
+                match maybe_key {
+                    Some(key) => {
+                        if !try_send(target, key, &mut retry_interval, &mut done_ch, logger).await {
+                            return;
+                        }
+                    }
+                    None => return,
+                }
+            }
+            _ = done_ch.recv() => {
+                return;
+            }
+        }
+    }
+}
+
+// 流式传输项目
+pub async fn stream_items<T>(store: Arc<dyn Store<T>>, target: &dyn Target, done_ch: mpsc::Receiver<()>, logger: Logger)
+where
+    T: Serialize + DeserializeOwned + Send + Sync + 'static,
+{
+    // 创建一个 done_ch 的克隆，以便可以将其传递给 replay_items
+    // let (tx, rx) = mpsc::channel::<()>(1);
+
+    let (tx_replay, rx_replay) = mpsc::channel::<()>(1);
+    let (tx_send, rx_send) = mpsc::channel::<()>(1);
+
+    let mut done_ch = done_ch;
+
+    let key_ch = replay_items(store, rx_replay, logger, &target.name()).await;
+    // let key_ch = replay_items(store, rx, logger, &target.name()).await;
+
+    let tx_replay_clone = tx_replay.clone();
+    let tx_send_clone = tx_send.clone();
+
+    // 监听原始 done_ch，如果收到信号，则关闭我们创建的通道
+    tokio::spawn(async move {
+        // if done_ch.recv().await.is_some() {
+        //     let _ = tx.send(()).await;
+        // }
+        if done_ch.recv().await.is_some() {
+            let _ = tx_replay_clone.send(()).await;
+            let _ = tx_send_clone.send(()).await;
+        }
+    });
+
+    // send_items(target, key_ch, rx, logger).await;
+    send_items(target, key_ch, rx_send, logger).await;
+}

crates/notify/src/store/queue.rs

@@ -1,515 +1,243 @@
-use common::error::{Error, Result};
+use crate::store::{parse_key, Key, Store, StoreError, StoreResult, DEFAULT_EXT, DEFAULT_LIMIT};
+use async_trait::async_trait;
 use serde::{de::DeserializeOwned, Serialize};
 use snap::raw::{Decoder, Encoder};
-use std::collections::HashMap;
-use std::io::Read;
-use std::marker::PhantomData;
+use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
-use std::sync::{Arc, RwLock};
 use std::time::{SystemTime, UNIX_EPOCH};
-use std::{fs, io};
+use tokio::fs;
+use tokio::sync::RwLock;
 use uuid::Uuid;
 
-/// Keys in storage
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct Key {
-    /// Key name
-    pub name: String,
-    /// Whether to compress
-    pub compress: bool,
-    /// filename extension
-    pub extension: String,
-    /// Number of items
-    pub item_count: usize,
-}
-
-impl Key {
-    /// Create a new key
-    pub fn new(name: impl Into<String>, extension: impl Into<String>, compress: bool) -> Self {
-        Self {
-            name: name.into(),
-            compress,
-            extension: extension.into(),
-            item_count: 1,
-        }
-    }
-
-    /// Convert to string form
-    #[allow(clippy::inherent_to_string)]
-    pub fn to_string(&self) -> String {
-        let mut key_str = self.name.clone();
-        if self.item_count > 1 {
-            key_str = format!("{}:{}", self.item_count, self.name);
-        }
-
-        let compress_ext = if self.compress { COMPRESS_EXT } else { "" };
-        format!("{}{}{}", key_str, self.extension, compress_ext)
-    }
-}
-
-/// Parse key from file name
-#[allow(clippy::redundant_closure)]
-pub fn parse_key(filename: &str) -> Key {
-    let compress = filename.ends_with(COMPRESS_EXT);
-    let filename = if compress {
-        &filename[..filename.len() - 7] // 移除 ".snappy"
-    } else {
-        filename
-    };
-
-    let mut parts = filename.splitn(2, '.');
-    let name_part = parts.next().unwrap_or("");
-    let extension = parts
-        .next()
-        .map_or_else(|| String::new(), |ext| format!(".{}", ext))
-        .to_string();
-
-    let mut name = name_part.to_string();
-    let mut item_count = 1;
-
-    if let Some(pos) = name_part.find(':') {
-        if let Ok(count) = name_part[..pos].parse::<usize>() {
-            item_count = count;
-            name = name_part[pos + 1..].to_string();
-        }
-    }
-
-    Key {
-        name,
-        compress,
-        extension,
-        item_count,
-    }
-}
-
-/// Store the characteristics of the project
-pub trait Store<T>: Send + Sync
-where
-    T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
-{
-    /// Store a single item
-    fn put(&self, item: T) -> Result<Key>;
-
-    /// Store multiple projects
-    fn put_multiple(&self, items: Vec<T>) -> Result<Key>;
-
-    /// Get a single item
-    fn get(&self, key: &Key) -> Result<T>;
-
-    /// Get multiple items
-    fn get_multiple(&self, key: &Key) -> Result<Vec<T>>;
-
-    /// Get the raw bytes
-    fn get_raw(&self, key: &Key) -> Result<Vec<u8>>;
-
-    /// Stores raw bytes
-    fn put_raw(&self, data: &[u8]) -> Result<Key>;
-
-    /// Gets the number of items in storage
-    fn len(&self) -> usize;
-
-    /// Whether it is empty or not
-    fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
-
-    /// Lists all keys
-    fn list(&self) -> Vec<Key>;
-
-    /// Delete the key
-    fn del(&self, key: &Key) -> Result<()>;
-
-    /// Open Storage
-    fn open(&self) -> Result<()>;
-
-    /// Delete the storage
-    fn delete(&self) -> Result<()>;
-}
-
-const DEFAULT_LIMIT: u64 = 100000;
-const DEFAULT_EXT: &str = ".unknown";
-const COMPRESS_EXT: &str = ".snappy";
-
-/// Queue storage implementation
 pub struct QueueStore<T> {
-    /// Project Limitations
     entry_limit: u64,
-    /// Storage directory
     directory: PathBuf,
-    /// filename extension
     file_ext: String,
-    /// Item mapping: key -> modified time (Unix nanoseconds)
-    entries: Arc<RwLock<HashMap<String, i64>>>,
-    /// Type tags
-    _phantom: PhantomData<T>,
-    /// Whether to compress
-    compress: bool,
-    /// Store name
-    name: String,
+    entries: RwLock<BTreeMap<String, i64>>,
+    _phantom: std::marker::PhantomData<T>,
 }
 
 impl<T> QueueStore<T>
 where
-    T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    T: Serialize + DeserializeOwned + Send + Sync + 'static,
 {
-    /// Create a new queue store
-    pub fn new<P: AsRef<Path>>(directory: P, name: String, limit: u64, ext: Option<String>) -> Self {
-        let limit = if limit == 0 { DEFAULT_LIMIT } else { limit };
-        let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string());
-        let mut path = PathBuf::from(directory.as_ref());
-        path.push(&name);
-
-        // Create a directory (if it doesn't exist)
-        if !path.exists() {
-            if let Err(e) = fs::create_dir_all(&path) {
-                tracing::error!("创建存储目录失败 {}: {}", path.display(), e);
-            }
-        }
+    pub fn new<P: AsRef<Path>>(directory: P, limit: u64, ext: Option<&str>) -> Self {
+        let entry_limit = if limit == 0 { DEFAULT_LIMIT } else { limit };
+        let ext = ext.unwrap_or(DEFAULT_EXT).to_string();
 
         Self {
             directory: directory.as_ref().to_path_buf(),
-            name,
-            entry_limit: limit,
+            entry_limit,
             file_ext: ext,
-            compress: true, // Default to compressing
-            entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))),
-            _phantom: PhantomData,
+            entries: RwLock::new(BTreeMap::new()),
+            _phantom: std::marker::PhantomData,
         }
     }
 
-    /// Set the file extension
-    pub fn with_file_ext(mut self, file_ext: &str) -> Self {
-        self.file_ext = file_ext.to_string();
-        self
-    }
-
-    /// Set whether to compress or not
-    pub fn with_compression(mut self, compress: bool) -> Self {
-        self.compress = compress;
-        self
-    }
-
-    /// Get the file path
-    fn get_file_path(&self, key: &Key) -> PathBuf {
-        let mut filename = key.to_string();
-        filename.push_str(if self.compress { COMPRESS_EXT } else { &self.file_ext });
-        self.directory.join(filename)
-    }
-
-    /// Serialize the project
-    fn serialize_item(&self, item: &T) -> Result<Vec<u8>> {
-        let data = serde_json::to_vec(item).map_err(|e| Error::msg(format!("Serialization failed: {}", e)))?;
-
-        if self.compress {
-            let mut encoder = Encoder::new();
-            Ok(encoder
-                .compress_vec(&data)
-                .map_err(|e| Error::msg(format!("Compression failed: {}", e)))?)
-        } else {
-            Ok(data)
-        }
-    }
-
-    /// Deserialize the project
-    fn deserialize_item(&self, data: &[u8], is_compressed: bool) -> Result<T> {
-        let data = if is_compressed {
-            let mut decoder = Decoder::new();
-            decoder
-                .decompress_vec(data)
-                .map_err(|e| Error::msg(format!("Unzipping failed: {}", e)))?
-        } else {
-            data.to_vec()
-        };
-
-        serde_json::from_slice(&data).map_err(|e| Error::msg(format!("Deserialization failed: {}", e)))
-    }
-
-    /// Lists all files in the directory, sorted by modification time (oldest takes precedence.)）
-    fn list_files(&self) -> Result<Vec<fs::DirEntry>> {
-        let mut files = Vec::new();
-
-        for entry in fs::read_dir(&self.directory)? {
-            let entry = entry?;
-            let metadata = entry.metadata()?;
-            if metadata.is_file() {
-                files.push(entry);
-            }
-        }
-
-        // Sort by modification time
-        files.sort_by(|a, b| {
-            let a_time = a
-                .metadata()
-                .map(|m| m.modified())
-                .unwrap_or(Ok(UNIX_EPOCH))
-                .unwrap_or(UNIX_EPOCH);
-            let b_time = b
-                .metadata()
-                .map(|m| m.modified())
-                .unwrap_or(Ok(UNIX_EPOCH))
-                .unwrap_or(UNIX_EPOCH);
-            a_time.cmp(&b_time)
-        });
-
-        Ok(files)
-    }
-
-    /// Write the object to a file
-    fn write_object(&self, key: &Key, item: &T) -> Result<()> {
-        // Serialize the object
-        let data = serde_json::to_vec(item)?;
-        self.write_bytes(key, &data)
-    }
-
-    /// Write multiple objects to a file
-    fn write_multiple_objects(&self, key: &Key, items: &[T]) -> Result<()> {
-        let mut data = Vec::new();
-        for item in items {
-            let item_data = serde_json::to_vec(item)?;
-            data.extend_from_slice(&item_data);
-            data.push(b'\n');
-        }
-        self.write_bytes(key, &data)
-    }
-
-    /// Write bytes to a file
-    fn write_bytes(&self, key: &Key, data: &[u8]) -> Result<()> {
+    async fn write_bytes(&self, key: Key, data: Vec<u8>) -> StoreResult<()> {
         let path = self.directory.join(key.to_string());
 
-        let file_data = if key.compress {
-            // Use snap to compress data
+        let data = if key.compress {
             let mut encoder = Encoder::new();
-            encoder
-                .compress_vec(data)
-                .map_err(|e| Error::msg(format!("Compression failed:{}", e)))?
+            encoder.compress_vec(&data).map_err(|e| StoreError::Other(e.to_string()))?
         } else {
-            data.to_vec()
+            data
         };
 
-        // Make sure the directory exists
-        if let Some(parent) = path.parent() {
-            fs::create_dir_all(parent)?;
-        }
+        fs::write(&path, &data).await?;
 
-        // Write to the file
-        fs::write(&path, &file_data)?;
+        // 更新条目映射
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|e| StoreError::Other(e.to_string()))?
+            .as_nanos() as i64;
 
-        // Update the item mapping
-        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
-
-        let mut entries = self
-            .entries
-            .write()
-            .map_err(|_| Error::msg("Failed to obtain a write lock"))?;
-        entries.insert(key.to_string(), now);
+        self.entries.write().await.insert(key.to_string(), now);
 
         Ok(())
     }
 
-    /// Read bytes from a file
-    fn read_bytes(&self, key: &Key) -> Result<Vec<u8>> {
-        let path = self.directory.join(key.to_string());
-        let data = fs::read(&path)?;
-
-        if data.is_empty() {
-            return Err(Error::msg("The file is empty"));
-        }
-
-        if key.compress {
-            // Use Snap to extract the data
-            let mut decoder = Decoder::new();
-            decoder
-                .decompress_vec(&data)
-                .map_err(|e| Error::msg(format!("Failed to decompress:{}", e)))
-        } else {
-            Ok(data)
-        }
+    async fn write(&self, key: Key, item: T) -> StoreResult<()> {
+        let data = serde_json::to_vec(&item)?;
+        self.write_bytes(key, data).await
     }
 
-    /// Check whether the storage limit is reached
-    fn check_entry_limit(&self) -> Result<()> {
-        let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?;
-        if entries.len() as u64 >= self.entry_limit {
-            return Err(Error::msg("The storage limit has been reached"));
+    async fn multi_write(&self, key: Key, items: Vec<T>) -> StoreResult<()> {
+        let mut buffer = Vec::new();
+
+        for item in items {
+            let item_data = serde_json::to_vec(&item)?;
+            buffer.extend_from_slice(&item_data);
+            buffer.push(b'\n'); // 使用换行符分隔项目
         }
+
+        self.write_bytes(key, buffer).await
+    }
+
+    async fn del_internal(&self, key: &Key) -> StoreResult<()> {
+        let path = self.directory.join(key.to_string());
+
+        if let Err(e) = fs::remove_file(&path).await {
+            if e.kind() != std::io::ErrorKind::NotFound {
+                return Err(e.into());
+            }
+        }
+
+        self.entries.write().await.remove(&key.to_string());
+
         Ok(())
     }
 }
 
+#[async_trait]
 impl<T> Store<T> for QueueStore<T>
 where
-    T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
+    T: Serialize + DeserializeOwned + Send + Sync + 'static,
 {
-    fn put(&self, item: T) -> Result<Key> {
-        {
-            self.check_entry_limit()?;
+    async fn put(&self, item: T) -> StoreResult<Key> {
+        let entries_len = self.entries.read().await.len() as u64;
+        if entries_len >= self.entry_limit {
+            return Err(StoreError::LimitExceeded);
         }
 
-        // generate a new uuid
+        // 生成 UUID 作为键
         let uuid = Uuid::new_v4();
-        let key = Key::new(uuid.to_string(), &self.file_ext, true);
+        let key = Key::new(uuid.to_string(), self.file_ext.clone());
 
-        self.write_object(&key, &item)?;
-
-        Ok(key)
-    }
-    fn put_multiple(&self, items: Vec<T>) -> Result<Key> {
-        if items.is_empty() {
-            return Err(Error::msg("The list of items is empty"));
-        }
-
-        {
-            self.check_entry_limit()?;
-        }
-
-        // Generate a new UUID
-        let uuid = Uuid::new_v4();
-        let mut key = Key::new(uuid.to_string(), &self.file_ext, true);
-        key.item_count = items.len();
-
-        self.write_multiple_objects(&key, &items)?;
+        self.write(key.clone(), item).await?;
 
         Ok(key)
     }
 
-    fn get(&self, key: &Key) -> Result<T> {
-        let items = self.get_multiple(key)?;
-        if items.is_empty() {
-            return Err(Error::msg("Item not found"));
+    async fn put_multiple(&self, items: Vec<T>) -> StoreResult<Key> {
+        let entries_len = self.entries.read().await.len() as u64;
+        if entries_len >= self.entry_limit {
+            return Err(StoreError::LimitExceeded);
         }
 
-        Ok(items[0].clone())
+        if items.is_empty() {
+            return Err(StoreError::Other("Cannot store empty item list".into()));
+        }
+
+        // 生成 UUID 作为键
+        let uuid = Uuid::new_v4();
+        let key = Key::new(uuid.to_string(), self.file_ext.clone())
+            .with_item_count(items.len())
+            .with_compression(true);
+
+        self.multi_write(key.clone(), items).await?;
+
+        Ok(key)
     }
 
-    fn get_multiple(&self, key: &Key) -> Result<Vec<T>> {
-        let data = self.get_raw(key)?;
+    async fn get(&self, key: Key) -> StoreResult<T> {
+        let items = self.get_multiple(key).await?;
+        items
+            .into_iter()
+            .next()
+            .ok_or_else(|| StoreError::Other("No items found".into()))
+    }
 
-        let mut items = Vec::with_capacity(key.item_count);
-        let mut reader = io::Cursor::new(&data);
+    async fn get_multiple(&self, key: Key) -> StoreResult<Vec<T>> {
+        let data = self.get_raw(key).await?;
 
-        // Try to read each JSON object
-        let mut buffer = Vec::new();
-
-        // if the read fails try parsing it once
-        if reader.read_to_end(&mut buffer).is_err() {
-            // Try to parse the entire data as a single object
-            return match serde_json::from_slice::<T>(&data) {
-                Ok(item) => {
-                    items.push(item);
-                    Ok(items)
-                }
-                Err(_) => {
-                    // An attempt was made to resolve to an array of objects
-                    match serde_json::from_slice::<Vec<T>>(&data) {
-                        Ok(array_items) => Ok(array_items),
-                        Err(e) => Err(Error::msg(format!("Failed to parse the data:{}", e))),
-                    }
-                }
-            };
+        // 尝试解析为 JSON 数组
+        match serde_json::from_slice::<Vec<T>>(&data) {
+            Ok(items) if !items.is_empty() => return Ok(items),
+            Ok(_) => return Err(StoreError::Other("No items deserialized".into())),
+            Err(_) => {} // 失败则尝试按行解析
         }
-
-        // Read JSON objects by row
-        for line in buffer.split(|&b| b == b'\n') {
-            if !line.is_empty() {
-                match serde_json::from_slice::<T>(line) {
-                    Ok(item) => items.push(item),
-                    Err(e) => tracing::warn!("Failed to parse row data:{}", e),
-                }
+        // 如果直接解析为 Vec<T> 失败，则尝试按行解析
+        // 转换为字符串并按行解析
+        let data_str = std::str::from_utf8(&data).map_err(StoreError::Utf8)?;
+        // 按行解析（JSON Lines）
+        let mut items = Vec::new();
+        for line in data_str.lines() {
+            let line = line.trim();
+            if line.is_empty() {
+                continue;
             }
+            let item = serde_json::from_str::<T>(line).map_err(StoreError::Deserialize)?;
+            items.push(item);
         }
 
         if items.is_empty() {
-            return Err(Error::msg("Failed to resolve any items"));
+            return Err(StoreError::Other("Failed to deserialize items".into()));
         }
 
         Ok(items)
     }
 
-    fn get_raw(&self, key: &Key) -> Result<Vec<u8>> {
-        let data = self.read_bytes(key)?;
+    async fn get_raw(&self, key: Key) -> StoreResult<Vec<u8>> {
+        let path = self.directory.join(key.to_string());
+        let data = fs::read(&path).await?;
 
-        // Delete the wrong file
         if data.is_empty() {
-            let _ = self.del(key);
-            return Err(Error::msg("the file is empty"));
+            return Err(StoreError::Other("Empty file".into()));
         }
 
-        Ok(data)
+        if key.compress {
+            let mut decoder = Decoder::new();
+            decoder.decompress_vec(&data).map_err(|e| StoreError::Other(e.to_string()))
+        } else {
+            Ok(data)
+        }
     }
 
-    fn put_raw(&self, data: &[u8]) -> Result<Key> {
-        {
-            let entries = self.entries.read().map_err(|_| Error::msg("Failed to obtain a read lock"))?;
-            if entries.len() as u64 >= self.entry_limit {
-                return Err(Error::msg("the storage limit has been reached"));
-            }
+    async fn put_raw(&self, data: Vec<u8>) -> StoreResult<Key> {
+        let entries_len = self.entries.read().await.len() as u64;
+        if entries_len >= self.entry_limit {
+            return Err(StoreError::LimitExceeded);
         }
 
-        // Generate a new UUID
+        // 生成 UUID 作为键
         let uuid = Uuid::new_v4();
-        let key = Key::new(uuid.to_string(), &self.file_ext, true);
+        let key = Key::new(uuid.to_string(), self.file_ext.clone());
 
-        self.write_bytes(&key, data)?;
+        self.write_bytes(key.clone(), data).await?;
 
         Ok(key)
     }
 
-    fn len(&self) -> usize {
-        self.entries.read().map(|e| e.len()).unwrap_or(0)
+    async fn len(&self) -> usize {
+        self.entries.read().await.len()
     }
 
-    fn list(&self) -> Vec<Key> {
-        let entries = match self.entries.read() {
-            Ok(guard) => guard,
-            Err(_) => return Vec::new(),
-        };
+    async fn list(&self) -> Vec<Key> {
+        let entries = self.entries.read().await;
 
-        // Convert entries to vectors and sort by timestamp
-        let mut entries_vec: Vec<_> = entries.iter().collect();
-        entries_vec.sort_by(|a, b| a.1.cmp(b.1));
+        // 将条目转换为 (key, timestamp) 元组并排序
+        let mut entries_vec: Vec<(&String, &i64)> = entries.iter().collect();
+        entries_vec.sort_by_key(|(_k, &v)| v);
 
-        // Parsing key
-        entries_vec.iter().map(|(filename, _)| parse_key(filename)).collect()
+        // 将排序后的键解析为 Key 结构体
+        entries_vec.into_iter().map(|(k, _)| parse_key(k)).collect()
     }
 
-    fn del(&self, key: &Key) -> Result<()> {
-        let path = self.directory.join(key.to_string());
-
-        // Delete the file
-        if let Err(e) = fs::remove_file(&path) {
-            if e.kind() != io::ErrorKind::NotFound {
-                return Err(e.into());
-            }
-        }
-
-        // Remove the entry from the map
-        let mut entries = self
-            .entries
-            .write()
-            .map_err(|_| Error::msg("Failed to obtain a write lock"))?;
-        entries.remove(&key.to_string());
-
-        Ok(())
+    async fn del(&self, key: Key) -> StoreResult<()> {
+        self.del_internal(&key).await
     }
 
-    fn open(&self) -> Result<()> {
-        // Create a directory (if it doesn't exist)
-        fs::create_dir_all(&self.directory)?;
+    async fn open(&self) -> StoreResult<()> {
+        // 创建目录（如果不存在）
+        fs::create_dir_all(&self.directory).await?;
 
-        // Read existing files
-        let files = self.list_files()?;
+        // 读取已经存在的文件
+        let entries = self.entries.write();
+        let mut entries = entries.await;
+        entries.clear();
 
-        let mut entries = self
-            .entries
-            .write()
-            .map_err(|_| Error::msg("Failed to obtain a write lock"))?;
+        let mut dir_entries = fs::read_dir(&self.directory).await?;
+        while let Some(entry) = dir_entries.next_entry().await? {
+            if let Ok(metadata) = entry.metadata().await {
+                if metadata.is_file() {
+                    let modified = metadata
+                        .modified()?
+                        .duration_since(UNIX_EPOCH)
+                        .map_err(|e| StoreError::Other(e.to_string()))?
+                        .as_nanos() as i64;
 
-        for file in files {
-            if let Ok(meta) = file.metadata() {
-                if let Ok(modified) = meta.modified() {
-                    if let Ok(since_epoch) = modified.duration_since(UNIX_EPOCH) {
-                        entries.insert(file.file_name().to_string_lossy().to_string(), since_epoch.as_nanos() as i64);
-                    }
+                    entries.insert(entry.file_name().to_string_lossy().to_string(), modified);
                 }
             }
         }
@@ -517,8 +245,8 @@ where
         Ok(())
     }
 
-    fn delete(&self) -> Result<()> {
-        fs::remove_dir_all(&self.directory)?;
+    async fn delete(&self) -> StoreResult<()> {
+        fs::remove_dir_all(&self.directory).await?;
         Ok(())
     }
 }

crates/notify/src/system.rs

@@ -1,4 +1,4 @@
-use crate::config::notifier::EventNotifierConfig;
+use crate::config::EventNotifierConfig;
 use crate::notifier::EventNotifier;
 use common::error::Result;
 use ecstore::store::ECStore;