diff --git a/.docker/mqtt/config/emqx.conf b/.docker/mqtt/config/emqx.conf new file mode 100644 index 000000000..367f2296b --- /dev/null +++ b/.docker/mqtt/config/emqx.conf @@ -0,0 +1,37 @@ +# 节点配置 +node.name = "emqx@127.0.0.1" +node.cookie = "aBcDeFgHiJkLmNoPqRsTuVwXyZ012345" +node.data_dir = "/opt/emqx/data" + +# 日志配置 +log.console = {level = info, enable = true} +log.file = {path = "/opt/emqx/log/emqx.log", enable = true, level = info} + +# MQTT TCP 监听器 +listeners.tcp.default = {bind = "0.0.0.0:1883", max_connections = 1000000, enable = true} + +# MQTT SSL 监听器 +listeners.ssl.default = {bind = "0.0.0.0:8883", enable = false} + +# MQTT WebSocket 监听器 +listeners.ws.default = {bind = "0.0.0.0:8083", enable = true} + +# MQTT WebSocket SSL 监听器 +listeners.wss.default = {bind = "0.0.0.0:8084", enable = false} + +# 管理控制台 +dashboard.listeners.http = {bind = "0.0.0.0:18083", enable = true} + +# HTTP API +management.listeners.http = {bind = "0.0.0.0:8081", enable = true} + +# 认证配置 +authentication = [ + {enable = true, mechanism = password_based, backend = built_in_database, user_id_type = username} +] + +# 授权配置 +authorization.sources = [{type = built_in_database, enable = true}] + +# 持久化消息存储 +message.storage.backend = built_in_database \ No newline at end of file diff --git a/.docker/mqtt/config/vm.args b/.docker/mqtt/config/vm.args new file mode 100644 index 000000000..3ddbb9594 --- /dev/null +++ b/.docker/mqtt/config/vm.args @@ -0,0 +1,9 @@ +-name emqx@127.0.0.1 +-setcookie aBcDeFgHiJkLmNoPqRsTuVwXyZ012345 ++P 2097152 ++t 1048576 ++zdbbl 32768 +-kernel inet_dist_listen_min 6000 +-kernel inet_dist_listen_max 6100 +-smp enable +-mnesia dir "/opt/emqx/data/mnesia" \ No newline at end of file diff --git a/.docker/mqtt/docker-compose-more.yml b/.docker/mqtt/docker-compose-more.yml new file mode 100644 index 000000000..16128ab16 --- /dev/null +++ b/.docker/mqtt/docker-compose-more.yml @@ -0,0 +1,60 @@ +services: + emqx: + image: emqx/emqx:latest + container_name: emqx + restart: unless-stopped + environment: + - EMQX_NODE__NAME=emqx@127.0.0.1 + - EMQX_NODE__COOKIE=aBcDeFgHiJkLmNoPqRsTuVwXyZ012345 + - EMQX_NODE__DATA_DIR=/opt/emqx/data + - EMQX_LOG__CONSOLE__LEVEL=info + - EMQX_LOG__CONSOLE__ENABLE=true + - EMQX_LOG__FILE__PATH=/opt/emqx/log/emqx.log + - EMQX_LOG__FILE__LEVEL=info + - EMQX_LOG__FILE__ENABLE=true + - EMQX_LISTENERS__TCP__DEFAULT__BIND=0.0.0.0:1883 + - EMQX_LISTENERS__TCP__DEFAULT__MAX_CONNECTIONS=1000000 + - EMQX_LISTENERS__TCP__DEFAULT__ENABLE=true + - EMQX_LISTENERS__SSL__DEFAULT__BIND=0.0.0.0:8883 + - EMQX_LISTENERS__SSL__DEFAULT__ENABLE=false + - EMQX_LISTENERS__WS__DEFAULT__BIND=0.0.0.0:8083 + - EMQX_LISTENERS__WS__DEFAULT__ENABLE=true + - EMQX_LISTENERS__WSS__DEFAULT__BIND=0.0.0.0:8084 + - EMQX_LISTENERS__WSS__DEFAULT__ENABLE=false + - EMQX_DASHBOARD__LISTENERS__HTTP__BIND=0.0.0.0:18083 + - EMQX_DASHBOARD__LISTENERS__HTTP__ENABLE=true + - EMQX_MANAGEMENT__LISTENERS__HTTP__BIND=0.0.0.0:8081 + - EMQX_MANAGEMENT__LISTENERS__HTTP__ENABLE=true + - EMQX_AUTHENTICATION__1__ENABLE=true + - EMQX_AUTHENTICATION__1__MECHANISM=password_based + - EMQX_AUTHENTICATION__1__BACKEND=built_in_database + - EMQX_AUTHENTICATION__1__USER_ID_TYPE=username + - EMQX_AUTHORIZATION__SOURCES__1__TYPE=built_in_database + - EMQX_AUTHORIZATION__SOURCES__1__ENABLE=true + ports: + - "1883:1883" # MQTT TCP + - "8883:8883" # MQTT SSL + - "8083:8083" # MQTT WebSocket + - "8084:8084" # MQTT WebSocket SSL + - "18083:18083" # Web 管理控制台 + - "8081:8081" # HTTP API + volumes: + - ./data:/opt/emqx/data + - ./log:/opt/emqx/log + - ./config:/opt/emqx/etc + networks: + - mqtt-net + healthcheck: + test: [ "CMD", "/opt/emqx/bin/emqx_ctl", "status" ] + interval: 30s + timeout: 10s + retries: 3 + logging: + driver: "json-file" + options: + max-size: "100m" + max-file: "3" + +networks: + mqtt-net: + driver: bridge diff --git a/.docker/mqtt/docker-compose.yml b/.docker/mqtt/docker-compose.yml new file mode 100644 index 000000000..040b76846 --- /dev/null +++ b/.docker/mqtt/docker-compose.yml @@ -0,0 +1,15 @@ +services: + emqx: + image: emqx/emqx:latest + container_name: emqx + ports: + - "1883:1883" + - "8083:8083" + - "8084:8084" + - "8883:8883" + - "18083:18083" + restart: unless-stopped + +networks: + default: + driver: bridge \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index 25298d122..c2005b764 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1664,9 +1664,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.39" +version = "4.5.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd60e63e9be68e5fb56422e397cf9baddded06dae1d2e523401542383bc72a9f" +checksum = "40b6887a1d8685cebccf115538db5c0efe625ccac9696ad45c409d96566e910f" dependencies = [ "clap_builder", "clap_derive", @@ -1674,9 +1674,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.5.39" +version = "4.5.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89cc6392a1f72bbeb820d71f32108f61fdaf18bc526e1d23954168a67759ef51" +checksum = "e0c66c08ce9f0c698cbce5c0279d0bb6ac936d8674174fe48f736533b964f59e" dependencies = [ "anstream", "anstyle", @@ -1686,9 +1686,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.5.32" +version = "4.5.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09176aae279615badda0765c0c0b3f6ed53f4709118af73cf4655d85d1530cd7" +checksum = "d2c7947ae4cc3d851207c1adb5b5e260ff0cca11446b1d6d1423788e442257ce" dependencies = [ "heck 0.5.0", "proc-macro2", @@ -8130,11 +8130,14 @@ dependencies = [ name = "rustfs-event" version = "0.0.1" dependencies = [ + "async-trait", "common", "ecstore", "once_cell", "reqwest", + "rumqttc", "rustfs-config", + "rustfs-notify", "serde", "serde_json", "serde_with", @@ -8144,6 +8147,8 @@ dependencies = [ "tokio", "tokio-util", "tracing", + "url", + "urlencoding", "uuid", ] diff --git a/crates/event/Cargo.toml b/crates/event/Cargo.toml index 9b9ddc681..e0348fcee 100644 --- a/crates/event/Cargo.toml +++ b/crates/event/Cargo.toml @@ -8,10 +8,12 @@ version.workspace = true [dependencies] rustfs-config = { workspace = true, features = ["constants", "notify"] } +rustfs-notify = { workspace = true } +async-trait = { workspace = true } common = { workspace = true } ecstore = { workspace = true } once_cell = { workspace = true } -reqwest = { workspace = true, optional = true } +reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } serde_with = { workspace = true } @@ -22,6 +24,9 @@ thiserror = { workspace = true } tokio = { workspace = true, features = ["sync", "net", "macros", "signal", "rt-multi-thread"] } tokio-util = { workspace = true } uuid = { workspace = true, features = ["v4", "serde"] } +url = { workspace = true } +urlencoding = { workspace = true } +rumqttc = { workspace = true } [lints] workspace = true diff --git a/crates/event/src/error.rs b/crates/event/src/error.rs index 3d138298f..1f64050fa 100644 --- a/crates/event/src/error.rs +++ b/crates/event/src/error.rs @@ -12,14 +12,6 @@ pub enum Error { 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("Feature disabled: {0}")] @@ -43,361 +35,3 @@ impl 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::(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::(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::(); - // 错误类型应该相对紧凑，考虑到包含多种错误类型，96 字节是合理的 - assert!(size <= 128, "Error size should be reasonable, got {} bytes", size); - - // 测试 Option的大小 - let option_size = mem::size_of::>(); - assert!(option_size <= 136, "Option should be efficient, got {} bytes", option_size); - } - - #[test] - fn test_error_thread_safety() { - // 测试错误类型的线程安全性 - fn assert_send() {} - fn assert_sync() {} - - assert_send::(); - assert_sync::(); - } - - #[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()); - } - } -} diff --git a/crates/event/src/event.rs b/crates/event/src/event.rs index 2a4270be4..dafe976ee 100644 --- a/crates/event/src/event.rs +++ b/crates/event/src/event.rs @@ -1,5 +1,4 @@ -use crate::Error; -use reqwest::dns::Name; +use crate::error::Error; use serde::{Deserialize, Serialize}; use serde_with::{DeserializeFromStr, SerializeDisplay}; use smallvec::{smallvec, SmallVec}; diff --git a/crates/event/src/lib.rs b/crates/event/src/lib.rs index f6e8ea019..2877c6166 100644 --- a/crates/event/src/lib.rs +++ b/crates/event/src/lib.rs @@ -2,3 +2,4 @@ mod error; mod event; mod notifier; mod system; +mod target; diff --git a/crates/event/src/notifier.rs b/crates/event/src/notifier.rs index 7105f6563..d819cc7d1 100644 --- a/crates/event/src/notifier.rs +++ b/crates/event/src/notifier.rs @@ -1,7 +1,7 @@ -use crate::config::EventNotifierConfig; -use crate::event::Event; use common::error::{Error, Result}; use ecstore::store::ECStore; +use rustfs_notify::Event; +use rustfs_notify::EventNotifierConfig; use std::sync::Arc; use tokio::sync::{broadcast, mpsc}; use tokio_util::sync::CancellationToken; @@ -25,7 +25,7 @@ impl EventNotifier { /// Create a new event notifier #[instrument(skip_all)] pub async fn new(store: Arc) -> Result { - let manager = crate::store::manager::EventManager::new(store); + let manager = rustfs_notify::manager::EventManager::new(store); let manager = Arc::new(manager.await); diff --git a/crates/event/src/system.rs b/crates/event/src/system.rs index 77ebbb54b..beeda788c 100644 --- a/crates/event/src/system.rs +++ b/crates/event/src/system.rs @@ -2,6 +2,8 @@ use crate::notifier::EventNotifier; use common::error::Result; use ecstore::store::ECStore; use once_cell::sync::OnceCell; +use rustfs_notify::Event; +use rustfs_notify::EventNotifierConfig; use std::sync::{Arc, Mutex}; use tracing::{debug, error, info}; @@ -37,7 +39,7 @@ impl EventSystem { } /// Send events - pub async fn send_event(&self, event: crate::Event) -> Result<()> { + pub async fn send_event(&self, event: Event) -> Result<()> { let guard = self .notifier .lock() diff --git a/crates/event/src/target/mod.rs b/crates/event/src/target/mod.rs new file mode 100644 index 000000000..32fcd4a4f --- /dev/null +++ b/crates/event/src/target/mod.rs @@ -0,0 +1,49 @@ +use async_trait::async_trait; +use rustfs_notify::store::{Key, Store, StoreError, StoreResult}; +use serde::{de::DeserializeOwned, Serialize}; +use std::sync::Arc; + +pub mod mqtt; +pub mod webhook; + +pub const STORE_PREFIX: &str = "rustfs"; + +// Target 公共 trait，对应 Go 的 Target 接口 +#[async_trait] +pub trait Target: Send + Sync { + fn name(&self) -> String; + async fn send_from_store(&self, key: Key) -> StoreResult<()>; + async fn is_active(&self) -> StoreResult; + async fn close(&self) -> StoreResult<()>; +} + +// TargetID 结构体，用于唯一标识目标 +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct TargetID { + pub id: String, + pub name: String, +} + +impl TargetID { + pub fn new(id: &str, name: &str) -> Self { + Self { + id: id.to_owned(), + name: name.to_owned(), + } + } +} + +impl ToString for TargetID { + fn to_string(&self) -> String { + format!("{}:{}", self.name, self.id) + } +} + +// TargetStore 接口 +pub trait TargetStore { + fn store(&self) -> Option>> + where + T: Serialize + DeserializeOwned + Send + Sync + 'static; +} + +pub type Logger = fn(ctx: Option<&str>, err: StoreError, id: &str, err_kind: &[&dyn std::fmt::Display]); diff --git a/crates/event/src/target/mqtt.rs b/crates/event/src/target/mqtt.rs new file mode 100644 index 000000000..a28ce5b96 --- /dev/null +++ b/crates/event/src/target/mqtt.rs @@ -0,0 +1,426 @@ +use super::{Logger, Target, TargetID, TargetStore, STORE_PREFIX}; +use async_trait::async_trait; +use once_cell::sync::OnceCell; +use rumqttc::{AsyncClient, ConnectionError, Event as MqttEvent, MqttOptions, QoS, Transport}; +use rustfs_config::notify::mqtt::MQTTArgs; +use rustfs_notify::store; +use rustfs_notify::{ + store::{Key, Store, StoreError, StoreResult}, + Event, QueueStore, +}; +use serde::{de::DeserializeOwned, Serialize}; +use serde_json::json; +use std::{path::PathBuf, sync::Arc, time::Duration}; +use tokio::{ + sync::{mpsc, Mutex}, + task::JoinHandle, +}; +use url::Url; + +pub struct MQTTTarget { + init: OnceCell<()>, + id: TargetID, + args: MQTTArgs, + client: Option>>, + eventloop_handle: Option>, + store: Option>>, + logger: Logger, + cancel_tx: mpsc::Sender<()>, + connection_status: Arc>, +} + +impl MQTTTarget { + pub async fn new(id: &str, args: MQTTArgs, logger: Logger) -> Result { + // 创建取消通道 + let (cancel_tx, mut cancel_rx) = mpsc::channel(1); + let connection_status = Arc::new(Mutex::new(false)); + + // 创建队列存储（如果配置了） + let mut store = None; + if !args.queue_dir.is_empty() { + if args.qos == 0 { + return Err(StoreError::Other("QoS should be set to 1 or 2 if queueDir is set".to_string())); + } + + let queue_dir = PathBuf::from(&args.queue_dir).join(format!("{}-mqtt-{}", STORE_PREFIX, id)); + let queue_store = Arc::new(QueueStore::::new(queue_dir, args.queue_limit, Some(".event"))); + + queue_store.open().await?; + store = Some(queue_store.clone() as Arc>); + + // 设置事件流 + let status_clone = connection_status.clone(); + let logger_clone = logger; + let target_store = queue_store; + let args_clone = args.clone(); + let id_clone = id.to_string(); + let cancel_tx_clone = cancel_tx.clone(); + + tokio::spawn(async move { + let target = Arc::new(MQTTTargetWrapper { + id: TargetID::new(&id_clone, "mqtt"), + args: args_clone, + client: None, + logger: logger_clone, + cancel_tx: cancel_tx_clone, + connection_status: status_clone, + }); + + store::stream_items(target_store, target, cancel_rx, logger_clone).await; + }); + } + + Ok(Self { + init: OnceCell::new(), + id: TargetID::new(id, "mqtt"), + args, + client: None, + eventloop_handle: None, + store, + logger, + cancel_tx, + connection_status, + }) + } + + async fn initialize(&self) -> StoreResult<()> { + if self.init.get().is_some() { + return Ok(()); + } + + // 解析 MQTT broker 地址 + let broker_url = Url::parse(&self.args.broker).map_err(|e| StoreError::Other(format!("Invalid broker URL: {}", e)))?; + + let host = broker_url + .host_str() + .ok_or_else(|| StoreError::Other("Missing host in broker URL".into()))? + .to_string(); + + let port = broker_url.port().unwrap_or_else(|| { + match broker_url.scheme() { + "mqtt" => 1883, + "mqtts" | "ssl" | "tls" => 8883, + "ws" => 80, + "wss" => 443, + _ => 1883, // 默认 + } + }); + + // 创建客户端 ID + let client_id = format!( + "{:x}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_err(|e| StoreError::Other(e.to_string()))? + .as_nanos() + ); + + // 创建 MQTT 选项 + let mut mqtt_options = MqttOptions::new(client_id, host, port); + mqtt_options.set_clean_session(true); + mqtt_options.set_keep_alive(self.args.keep_alive); + mqtt_options.set_max_packet_size(100 * 1024); // 100KB + + // 设置重连间隔 + mqtt_options.set_connection_timeout(self.args.keep_alive.as_secs() as u16); + mqtt_options.set_max_reconnect_retry(10); // 最大重试次数 + mqtt_options.set_retry_interval(Duration::from_millis(100)); + + // 如果设置了用户名和密码 + if !self.args.username.is_empty() { + mqtt_options.set_credentials(&self.args.username, &self.args.password); + } + + // TLS 配置 + if self.args.root_cas.is_some() + || broker_url.scheme() == "mqtts" + || broker_url.scheme() == "ssl" + || broker_url.scheme() == "tls" + || broker_url.scheme() == "wss" + { + let mut transport = if broker_url.scheme() == "ws" || broker_url.scheme() == "wss" { + let path = broker_url.path(); + Transport::Ws { + path: if path == "/" { "/mqtt".to_string() } else { path.to_string() }, + } + } else { + Transport::Tls + }; + + // 如果提供了根证书 + if let Some(root_cas) = &self.args.root_cas { + if let Transport::Tls = transport { + transport = Transport::Tls; + } + + // 在实际实现中，这里需要设置 TLS 证书 + // 由于 rumqttc 的接口可能会随版本变化，请参考最新的文档 + } + + mqtt_options.set_transport(transport); + } else if broker_url.scheme() == "ws" { + let path = broker_url.path(); + mqtt_options.set_transport(Transport::Ws { + path: if path == "/" { "/mqtt".to_string() } else { path.to_string() }, + }); + } + + // 创建 MQTT 客户端 + let (client, mut eventloop) = AsyncClient::new(mqtt_options, 10); + let client = Arc::new(Mutex::new(client)); + + // 克隆引用用于事件循环 + let connection_status = self.connection_status.clone(); + let client_clone = client.clone(); + let logger = self.logger; + let target_id = self.id.to_string(); + + // 启动事件循环 + let eventloop_handle = tokio::spawn(async move { + loop { + match eventloop.poll().await { + Ok(event) => match event { + MqttEvent::Incoming(incoming) => match incoming { + rumqttc::Packet::ConnAck(connack) => { + if connack.code == rumqttc::ConnectReturnCode::Success { + *connection_status.lock().await = true; + } else { + logger( + None, + StoreError::Other(format!("MQTT connection failed: {:?}", connack.code)), + &target_id, + &[], + ); + *connection_status.lock().await = false; + } + } + _ => {} + }, + MqttEvent::Outgoing(_) => {} + }, + Err(ConnectionError::ConnectionRefused(_)) => { + *connection_status.lock().await = false; + logger(None, StoreError::NotConnected, &target_id, &["MQTT connection refused"]); + tokio::time::sleep(Duration::from_secs(5)).await; + } + Err(e) => { + *connection_status.lock().await = false; + logger(None, StoreError::Other(format!("MQTT error: {}", e)), &target_id, &[]); + tokio::time::sleep(Duration::from_secs(5)).await; + } + } + } + }); + + // 更新目标状态 + self.client = Some(client_clone); + self.eventloop_handle = Some(eventloop_handle); + + // 等待连接建立 + for _ in 0..5 { + if *self.connection_status.lock().await { + self.init + .set(()) + .map_err(|_| StoreError::Other("Failed to initialize MQTT target".into()))?; + return Ok(()); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + + Err(StoreError::NotConnected) + } + + async fn send(&self, event_data: &Event) -> StoreResult<()> { + let client = match &self.client { + Some(client) => client, + None => return Err(StoreError::NotConnected), + }; + + if !*self.connection_status.lock().await { + return Err(StoreError::NotConnected); + } + + // 构建消息内容 + let object_key = urlencoding::decode(&event_data.s3.object.key) + .map_err(|e| StoreError::Other(format!("Failed to decode object key: {}", e)))?; + + let key = format!("{}/{}", event_data.s3.bucket.name, object_key); + let log_data = json!({ + "EventName": event_data.event_name, + "Key": key, + "Records": [event_data] + }); + + let payload = serde_json::to_string(&log_data).map_err(|e| StoreError::SerdeError(e))?; + + // 确定 QoS 级别 + let qos = match self.args.qos { + 0 => QoS::AtMostOnce, + 1 => QoS::AtLeastOnce, + 2 => QoS::ExactlyOnce, + _ => QoS::AtMostOnce, // 默认 + }; + + // 发布消息 + let mut client_guard = client.lock().await; + client_guard + .publish(&self.args.topic, qos, false, payload) + .await + .map_err(|e| { + if matches!(e, rumqttc::ClientError::ConnectionLost(_)) { + StoreError::NotConnected + } else { + StoreError::Other(format!("MQTT publish error: {}", e)) + } + })?; + + Ok(()) + } +} + +// MQTT 目标包装器，用于流事件 +struct MQTTTargetWrapper { + id: TargetID, + args: MQTTArgs, + client: Option>>, + logger: Logger, + cancel_tx: mpsc::Sender<()>, + connection_status: Arc>, +} + +#[async_trait] +impl Target for MQTTTargetWrapper { + fn name(&self) -> String { + self.id.to_string() + } + + async fn send_from_store(&self, _key: Key) -> StoreResult<()> { + // 这个方法在实际 MQTTTarget 中实现 + Ok(()) + } + + async fn is_active(&self) -> StoreResult { + Ok(*self.connection_status.lock().await) + } + + async fn close(&self) -> StoreResult<()> { + // 发送取消信号 + let _ = self.cancel_tx.send(()).await; + Ok(()) + } +} + +#[async_trait] +impl Target for MQTTTarget { + fn name(&self) -> String { + self.id.to_string() + } + + async fn send_from_store(&self, key: Key) -> StoreResult<()> { + self.initialize().await?; + + // 如果没有连接，返回错误 + if !*self.connection_status.lock().await { + return Err(StoreError::NotConnected); + } + + // 如果有存储，获取事件并发送 + if let Some(store) = &self.store { + match store.get(key.clone()).await { + Ok(event_data) => { + match self.send(&event_data).await { + Ok(_) => { + // 成功发送后删除事件 + return store.del(key).await.map(|_| ()); + } + Err(e) => { + (self.logger)(None, e.clone(), &self.id.to_string(), &["Failed to send event"]); + return Err(e); + } + } + } + Err(e) => { + // 如果文件不存在，忽略错误（可能已被处理） + if let StoreError::IOError(ref io_err) = e { + if io_err.kind() == std::io::ErrorKind::NotFound { + return Ok(()); + } + } + return Err(e); + } + } + } + + Ok(()) + } + + async fn is_active(&self) -> StoreResult { + if self.init.get().is_none() { + return Ok(false); + } + Ok(*self.connection_status.lock().await) + } + + async fn close(&self) -> StoreResult<()> { + // 发送取消信号 + let _ = self.cancel_tx.send(()).await; + + // 取消事件循环 + if let Some(handle) = &self.eventloop_handle { + handle.abort(); + } + + // 断开 MQTT 连接 + if let Some(client) = &self.client { + if let Ok(mut client) = client.try_lock() { + // 尝试断开连接（忽略错误） + let _ = client.disconnect().await; + } + } + + Ok(()) + } +} + +impl TargetStore for MQTTTarget { + fn store(&self) -> Option>> + where + T: Serialize + DeserializeOwned + Send + Sync + 'static, + { + if let Some(store) = &self.store { + // 类型检查确保 T 是 Event 类型 + if std::any::TypeId::of::() == std::any::TypeId::of::() { + // 安全：我们已经检查类型 ID 匹配 + let store_ptr = Arc::as_ptr(store); + let store_t = unsafe { Arc::from_raw(store_ptr as *const dyn Store) }; + // 增加引用计数，避免释放原始指针 + std::mem::forget(store_t.clone()); + return Some(store_t); + } + } + None + } +} + +impl MQTTTarget { + pub async fn save(&self, event_data: Event) -> StoreResult<()> { + // 如果配置了存储，则存储事件 + if let Some(store) = &self.store { + return store.put(event_data).await.map(|_| ()); + } + + // 否则，初始化并直接发送 + self.initialize().await?; + + // 检查连接 + if !*self.connection_status.lock().await { + return Err(StoreError::NotConnected); + } + + self.send(&event_data).await + } + + pub fn id(&self) -> &TargetID { + &self.id + } +} diff --git a/crates/event/src/target/webhook.rs b/crates/event/src/target/webhook.rs new file mode 100644 index 000000000..90a08e320 --- /dev/null +++ b/crates/event/src/target/webhook.rs @@ -0,0 +1,328 @@ +use super::{Logger, Target, TargetID, TargetStore, STORE_PREFIX}; +use async_trait::async_trait; +use once_cell::sync::OnceCell; +use reqwest::{header, Client, StatusCode}; +use rustfs_config::notify::webhook::WebhookArgs; +use rustfs_notify::{ + store::{self, Key, Store, StoreError, StoreResult}, + Event, +}; +use serde::de::DeserializeOwned; +use serde::Serialize; +use serde_json::json; +use std::{path::PathBuf, sync::Arc, time::Duration}; +use tokio::{net::TcpStream, sync::mpsc}; +use url::Url; + +pub struct WebhookTarget { + init: OnceCell<()>, + id: TargetID, + args: WebhookArgs, + client: Client, + store: Option>>, + logger: Logger, + cancel_tx: mpsc::Sender<()>, + addr: String, // 完整地址，包含 IP/DNS 和端口号 +} + +impl WebhookTarget { + pub async fn new(id: &str, args: WebhookArgs, logger: Logger) -> Result { + // 创建取消通道 + let (cancel_tx, cancel_rx) = mpsc::channel(1); + + // 配置客户端 + let mut client_builder = Client::builder().timeout(Duration::from_secs(10)); + + // 添加客户端证书如果配置了 + if !args.client_cert.is_empty() && !args.client_key.is_empty() { + let cert = + std::fs::read(&args.client_cert).map_err(|e| StoreError::Other(format!("Failed to read client cert: {}", e)))?; + let key = + std::fs::read(&args.client_key).map_err(|e| StoreError::Other(format!("Failed to read client key: {}", e)))?; + + let identity = reqwest::Identity::from_pem(&[cert, key].concat()) + .map_err(|e| StoreError::Other(format!("Failed to create identity: {}", e)))?; + + client_builder = client_builder.identity(identity); + } + + let client = client_builder + .build() + .map_err(|e| StoreError::Other(format!("Failed to create HTTP client: {}", e)))?; + + // 计算目标地址 + let endpoint = Url::parse(&args.endpoint).map_err(|e| StoreError::Other(format!("Invalid URL: {}", e)))?; + + let mut addr = endpoint + .host_str() + .ok_or_else(|| StoreError::Other("Missing host in endpoint".into()))? + .to_string(); + + // 如果没有端口，根据协议添加默认端口 + if endpoint.port().is_none() { + match endpoint.scheme() { + "http" => addr.push_str(":80"), + "https" => addr.push_str(":443"), + _ => return Err(StoreError::Other("Unsupported scheme".into())), + } + } else if let Some(port) = endpoint.port() { + addr = format!("{}:{}", addr, port); + } + + // 创建队列存储（如果配置了） + let mut store = None; + if !args.queue_dir.is_empty() { + let queue_dir = PathBuf::from(&args.queue_dir).join(format!("{}-webhook-{}", STORE_PREFIX, id)); + let queue_store = Arc::new(store::queue::QueueStore::::new(queue_dir, args.queue_limit, Some(".event"))); + + queue_store.open().await?; + store = Some(queue_store.clone() as Arc>); + + // 设置事件流 + let target_store = Arc::new(queue_store); + let target = Arc::new(WebhookTargetWrapper::new( + id, + args.clone(), + client.clone(), + addr.clone(), + logger, + cancel_tx.clone(), + )); + + tokio::spawn(async move { + store::stream_items(target_store.clone(), target.clone(), cancel_rx, logger).await; + }); + } + + Ok(Self { + init: OnceCell::new(), + id: TargetID::new(id, "webhook"), + args, + client, + store, + logger, + cancel_tx, + addr, + }) + } + + async fn initialize(&self) -> StoreResult<()> { + if self.init.get().is_some() { + return Ok(()); + } + + let is_active = self.is_active().await?; + if !is_active { + return Err(StoreError::NotConnected); + } + + self.init + .set(()) + .map_err(|_| StoreError::Other("Failed to initialize".into()))?; + Ok(()) + } + + async fn send(&self, event_data: &Event) -> StoreResult<()> { + // 构建请求数据 + let object_key = match urlencoding::decode(&event_data.s3.object.key) { + Ok(key) => key.to_string(), + Err(e) => return Err(StoreError::Other(format!("Failed to decode object key: {}", e))), + }; + + let key = format!("{}/{}", event_data.s3.bucket.name, object_key); + let log_data = json!({ + "EventName": event_data.event_name, + "Key": key, + "Records": [event_data] + }); + + // 创建请求 + let mut request_builder = self + .client + .post(&self.args.endpoint) + .header(header::CONTENT_TYPE, "application/json"); + + // 添加认证头 + if !self.args.auth_token.is_empty() { + let tokens: Vec<&str> = self.args.auth_token.split_whitespace().collect(); + match tokens.len() { + 2 => request_builder = request_builder.header(header::AUTHORIZATION, &self.args.auth_token), + 1 => request_builder = request_builder.header(header::AUTHORIZATION, format!("Bearer {}", &self.args.auth_token)), + _ => {} + } + } + + // 发送请求 + let response = request_builder.json(&log_data).send().await.map_err(|e| { + if e.is_timeout() || e.is_connect() { + StoreError::NotConnected + } else { + StoreError::Other(format!("Request failed: {}", e)) + } + })?; + + // 检查响应状态 + let status = response.status(); + if status.is_success() { + Ok(()) + } else if status == StatusCode::FORBIDDEN { + Err(StoreError::Other(format!( + "{} returned '{}', please check if your auth token is correctly set", + self.args.endpoint, status + ))) + } else { + Err(StoreError::Other(format!( + "{} returned '{}', please check your endpoint configuration", + self.args.endpoint, status + ))) + } + } +} + +struct WebhookTargetWrapper { + id: TargetID, + args: WebhookArgs, + client: Client, + addr: String, + logger: Logger, + cancel_tx: mpsc::Sender<()>, +} + +impl WebhookTargetWrapper { + fn new(id: &str, args: WebhookArgs, client: Client, addr: String, logger: Logger, cancel_tx: mpsc::Sender<()>) -> Self { + Self { + id: TargetID::new(id, "webhook"), + args, + client, + addr, + logger, + cancel_tx, + } + } +} + +#[async_trait] +impl Target for WebhookTargetWrapper { + fn name(&self) -> String { + self.id.to_string() + } + + async fn send_from_store(&self, key: Key) -> StoreResult<()> { + // 这个方法在 Target trait 实现中需要，但我们不会直接使用它 + // 实际上，它将由上面创建的 WebhookTarget 的 SendFromStore 方法处理 + Ok(()) + } + + async fn is_active(&self) -> StoreResult { + // 尝试连接到目标地址 + match tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&self.addr)).await { + Ok(Ok(_)) => Ok(true), + Ok(Err(e)) => { + if e.kind() == std::io::ErrorKind::ConnectionRefused + || e.kind() == std::io::ErrorKind::ConnectionAborted + || e.kind() == std::io::ErrorKind::ConnectionReset + { + Err(StoreError::NotConnected) + } else { + Err(StoreError::Other(format!("Connection error: {}", e))) + } + } + Err(_) => Err(StoreError::NotConnected), + } + } + + async fn close(&self) -> StoreResult<()> { + // 发送取消信号 + let _ = self.cancel_tx.send(()).await; + Ok(()) + } +} + +#[async_trait] +impl Target for WebhookTarget { + fn name(&self) -> String { + self.id.to_string() + } + + async fn send_from_store(&self, key: Key) -> StoreResult<()> { + self.initialize().await?; + + // 如果有存储，获取事件并发送 + if let Some(store) = &self.store { + match store.get(key.clone()).await { + Ok(event_data) => match self.send(&event_data).await { + Ok(_) => store.del(key).await?, + Err(e) => { + if matches!(e, StoreError::NotConnected) { + return Err(StoreError::NotConnected); + } + return Err(e); + } + }, + Err(e) => { + // 如果键不存在，可能已经被发送，忽略错误 + if let StoreError::IoError(io_err) = &e { + if io_err.kind() == std::io::ErrorKind::NotFound { + return Ok(()); + } + } + return Err(e); + } + } + } + + Ok(()) + } + + async fn is_active(&self) -> StoreResult { + // 尝试连接到目标地址 + match tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(&self.addr)).await { + Ok(Ok(_)) => Ok(true), + Ok(Err(_)) => Err(StoreError::NotConnected), + Err(_) => Err(StoreError::NotConnected), + } + } + + async fn close(&self) -> StoreResult<()> { + // 发送取消信号 + let _ = self.cancel_tx.send(()).await; + Ok(()) + } +} + +impl TargetStore for WebhookTarget { + fn store(&self) -> Option>> + where + T: Serialize + DeserializeOwned + Send + Sync + 'static, + { + if let Some(store) = &self.store { + // 注意：这里假设 T 是 Event 类型，需要类型转换（如果不是，将返回 None） + if std::any::TypeId::of::() == std::any::TypeId::of::() { + // 安全：因为我们检查了类型 ID + let store_ptr = Arc::as_ptr(store); + let store_t = unsafe { Arc::from_raw(store_ptr as *const dyn Store) }; + // 增加引用计数，避免释放原始指针 + std::mem::forget(store_t.clone()); + return Some(store_t); + } + } + None + } +} + +impl WebhookTarget { + pub async fn save(&self, event_data: Event) -> StoreResult<()> { + // 如果配置了存储，则存储事件 + if let Some(store) = &self.store { + return store.put(event_data).await.map(|_| ()); + } + + // 否则，初始化并直接发送 + self.initialize().await?; + self.send(&event_data).await + } + + pub fn id(&self) -> &TargetID { + &self.id + } +} diff --git a/crates/notify/src/adapter/mod.rs b/crates/notify/src/adapter/mod.rs index 7c2d9ee62..1eea53112 100644 --- a/crates/notify/src/adapter/mod.rs +++ b/crates/notify/src/adapter/mod.rs @@ -85,7 +85,7 @@ pub trait ChannelAdapter: Send + Sync + 'static { } /// Creates channel adapters based on the provided configuration. -pub fn create_adapters(configs: Vec) -> Result>, Error> { +pub async fn create_adapters(configs: Vec) -> Result>, Error> { let mut adapters: Vec> = Vec::new(); for config in configs { @@ -93,7 +93,7 @@ pub fn create_adapters(configs: Vec) -> Result { webhook_config.validate().map_err(Error::ConfigError)?; - adapters.push(Arc::new(webhook::WebhookAdapter::new(webhook_config.clone()))); + adapters.push(Arc::new(webhook::WebhookAdapter::new(webhook_config.clone()).await)); } #[cfg(feature = "mqtt")] AdapterConfig::Mqtt(mqtt_config) => { diff --git a/crates/notify/src/adapter/webhook.rs b/crates/notify/src/adapter/webhook.rs index 08f2df21c..6534e0c62 100644 --- a/crates/notify/src/adapter/webhook.rs +++ b/crates/notify/src/adapter/webhook.rs @@ -1,9 +1,12 @@ use crate::config::STORE_PREFIX; -use crate::{ChannelAdapter, ChannelAdapterType}; +use crate::error::Error; +use crate::store::Store; +use crate::{ChannelAdapter, ChannelAdapterType, QueueStore}; use crate::{Event, DEFAULT_RETRY_INTERVAL}; use async_trait::async_trait; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use reqwest::{self, Client, Identity, RequestBuilder}; +use rustfs_config::notify::webhook::WebhookArgs; use std::fs; use std::path::PathBuf; use std::sync::Arc; @@ -30,7 +33,7 @@ pub const ENV_WEBHOOK_CLIENT_KEY: &str = "RUSTFS_NOTIFY_WEBHOOK_CLIENT_KEY"; /// Webhook adapter for sending events to a webhook endpoint. pub struct WebhookAdapter { /// Configuration information - config: WebhookConfig, + config: WebhookArgs, /// Event storage queues store: Option>>, /// HTTP client @@ -39,16 +42,9 @@ pub struct WebhookAdapter { impl WebhookAdapter { /// Creates a new Webhook adapter. - pub fn new(config: WebhookConfig) -> Self { + pub async fn new(config: WebhookArgs) -> Self { let mut builder = Client::builder(); - if config.timeout.is_some() { - // Set the timeout for the client - match config.timeout { - Some(t) => builder = builder.timeout(Duration::from_secs(t)), - None => tracing::warn!("Timeout is not set, using default timeout"), - } - } - let client = if let (Some(cert_path), Some(key_path)) = (&config.client_cert, &config.client_key) { + let client = if let (cert_path, key_path) = (&config.client_cert, &config.client_key) { let cert_path = PathBuf::from(cert_path); let key_path = PathBuf::from(key_path); @@ -90,21 +86,20 @@ impl WebhookAdapter { }); // create a queue store if enabled - let store = if !config.common.queue_dir.len() > 0 { - let store_path = PathBuf::from(&config.common.queue_dir).join(format!( + let store = if !config.queue_dir.len() > 0 { + let store_path = PathBuf::from(&config.queue_dir).join(format!( "{}-{}-{}", STORE_PREFIX, Webhook.as_str(), - config.common.identifier + "identifier".to_string() )); - let queue_limit = if config.common.queue_limit > 0 { - config.common.queue_limit + let queue_limit = if config.queue_limit > 0 { + config.queue_limit } else { crate::config::default_queue_limit() }; - let name = config.common.identifier.clone(); - let store = QueueStore::new(store_path, name, queue_limit, Some(".event".to_string())); - if let Err(e) = store.open() { + let store = QueueStore::new(store_path, queue_limit, Some(".event")); + if let Err(e) = store.open().await { tracing::error!("Unable to open queue storage: {}", e); None } else { @@ -120,9 +115,10 @@ impl WebhookAdapter { /// Handle backlog events in storage pub async fn process_backlog(&self) -> Result<(), Error> { if let Some(store) = &self.store { - let keys = store.list(); + let keys = store.list().await; for key in keys { - match store.get_multiple(&key) { + let key_clone = key.clone(); + match store.get_multiple(key).await { Ok(events) => { for event in events { if let Err(e) = self.send_with_retry(&event).await { @@ -132,7 +128,7 @@ impl WebhookAdapter { } } // Deleted after successful processing - if let Err(e) = store.del(&key) { + if let Err(e) = store.del(key_clone).await { tracing::error!("Failed to delete a handled event: {}", e); } } @@ -140,7 +136,7 @@ impl WebhookAdapter { tracing::error!("Failed to read events from storage: {}", e); // delete the broken entries // If the event cannot be read, it may be corrupted, delete it - if let Err(del_err) = store.del(&key) { + if let Err(del_err) = store.del(key_clone).await { tracing::error!("Failed to delete a corrupted event: {}", del_err); } } @@ -153,10 +149,7 @@ impl WebhookAdapter { ///Send events to the webhook endpoint 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 - }; + let retry_interval = Duration::from_secs(DEFAULT_RETRY_INTERVAL); let mut attempts = 0; loop { @@ -164,18 +157,15 @@ impl WebhookAdapter { match self.send_request(event).await { Ok(_) => return Ok(()), Err(e) => { - if attempts <= self.config.max_retries { - tracing::warn!("Send to webhook fails and will be retried after 3 seconds:{}", e); - sleep(retry_interval).await; - } else if let Some(store) = &self.store { + tracing::warn!("Send to webhook fails and will be retried after 3 seconds:{}", e); + sleep(retry_interval).await; + if let Some(store) = &self.store { // store in a queue for later processing tracing::warn!("The maximum number of retries is reached, and the event is stored in a queue:{}", e); - if let Err(store_err) = store.put(event.clone()) { + if let Err(store_err) = store.put(event.clone()).await { tracing::error!("Events cannot be stored to a queue:{}", store_err); } return Err(e); - } else { - return Err(e); } } } @@ -211,7 +201,7 @@ impl WebhookAdapter { .post(&self.config.endpoint) .json(event) .header("Content-Type", "application/json"); - if let Some(token) = &self.config.auth_token { + if let token = &self.config.auth_token { let tokens: Vec<&str> = token.split_whitespace().collect(); match tokens.len() { 2 => request = request.header("Authorization", token), @@ -234,9 +224,10 @@ impl WebhookAdapter { /// 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)))?; + store.put(event.clone()).await.map_err(|e| { + tracing::error!("Failed to save event to queue: {}", e); + Error::Custom(format!("Failed to save event to queue: {}", e)) + })?; } Ok(()) } diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 5fdccafce..4f9a6d1ac 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -3,7 +3,7 @@ mod config; mod error; mod event; mod notifier; -mod store; +pub mod store; mod system; pub use adapter::create_adapters; @@ -17,3 +17,6 @@ pub use adapter::ChannelAdapterType; 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::manager; +pub use store::queue; +pub use store::queue::QueueStore; diff --git a/crates/notify/src/store/manager.rs b/crates/notify/src/store/manager.rs index 2573740b3..7355a09e4 100644 --- a/crates/notify/src/store/manager.rs +++ b/crates/notify/src/store/manager.rs @@ -157,7 +157,7 @@ impl EventManager { }; let adapter_configs = config.to_adapter_configs(); - match adapter::create_adapters(adapter_configs) { + match adapter::create_adapters(adapter_configs).await { Ok(adapters) => Ok(adapters), Err(err) => { tracing::error!("Failed to create adapters: {:?}", err); diff --git a/crates/notify/src/store/mod.rs b/crates/notify/src/store/mod.rs index 33ea11e0b..a4dd08f79 100644 --- a/crates/notify/src/store/mod.rs +++ b/crates/notify/src/store/mod.rs @@ -8,8 +8,8 @@ use std::time::Duration; use tokio::sync::mpsc; use tokio::time; -pub(crate) mod manager; -pub(crate) mod queue; +pub mod manager; +pub mod queue; // 常量定义 pub const RETRY_INTERVAL: Duration = Duration::from_secs(3); @@ -231,11 +231,16 @@ where } // 发送项目辅助函数 -pub async fn send_items(target: &dyn Target, mut key_ch: mpsc::Receiver, mut done_ch: mpsc::Receiver<()>, logger: Logger) { +pub async fn send_items( + target: Arc, + mut key_ch: mpsc::Receiver, + mut done_ch: mpsc::Receiver<()>, + logger: Logger, +) { let mut retry_interval = time::interval(RETRY_INTERVAL); - + let target_clone = target.clone(); async fn try_send( - target: &dyn Target, + target: Arc, key: Key, retry_interval: &mut time::Interval, done_ch: &mut mpsc::Receiver<()>, @@ -265,7 +270,7 @@ pub async fn send_items(target: &dyn Target, mut key_ch: mpsc::Receiver, mu maybe_key = key_ch.recv() => { match maybe_key { Some(key) => { - if !try_send(target, key, &mut retry_interval, &mut done_ch, logger).await { + if !try_send(target_clone.clone(), key, &mut retry_interval, &mut done_ch, logger).await { return; } } @@ -280,7 +285,7 @@ pub async fn send_items(target: &dyn Target, mut key_ch: mpsc::Receiver, mu } // 流式传输项目 -pub async fn stream_items(store: Arc>, target: &dyn Target, done_ch: mpsc::Receiver<()>, logger: Logger) +pub async fn stream_items(store: Arc>, target: Arc, done_ch: mpsc::Receiver<()>, logger: Logger) where T: Serialize + DeserializeOwned + Send + Sync + 'static, {