gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
houseme
2025-06-18 23:46:55 +08:00
parent 9339093638
commit 09e8bc8f02
19 changed files with 997 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/mqtt/config/emqx.conf
+37
View File
@@ -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
.docker/mqtt/config/vm.args
+9
View File
@@ -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"
.docker/mqtt/docker-compose-more.yml
+60
View File
@@ -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
.docker/mqtt/docker-compose.yml
+15
View File
@@ -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
Cargo.lock
Generated
+11 -6
View File
@@ -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",
]
crates/event/Cargo.toml
+6 -1
View File
@@ -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
crates/event/src/error.rs
-366
View File
@@ -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::SendError<crate::event::Event>>),
#[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::<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
+1 -2
View File
@@ -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};
crates/event/src/lib.rs
+1
View File
@@ -2,3 +2,4 @@ mod error;
mod event;
mod notifier;
mod system;
mod target;
crates/event/src/notifier.rs
+3 -3
View File
@@ -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<ECStore>) -> Result<Self> {
let manager = crate::store::manager::EventManager::new(store);
let manager = rustfs_notify::manager::EventManager::new(store);
let manager = Arc::new(manager.await);
crates/event/src/system.rs
+3 -1
View File
@@ -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()
crates/event/src/target/mod.rs
+49
View File
@@ -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<bool>;
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<T>(&self) -> Option<Arc<dyn Store<T>>>
where
T: Serialize + DeserializeOwned + Send + Sync + 'static;
}
pub type Logger = fn(ctx: Option<&str>, err: StoreError, id: &str, err_kind: &[&dyn std::fmt::Display]);
crates/event/src/target/mqtt.rs
+426
View File
@@ -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<Arc<Mutex<AsyncClient>>>,
eventloop_handle: Option<JoinHandle<()>>,
store: Option<Arc<dyn Store<Event>>>,
logger: Logger,
cancel_tx: mpsc::Sender<()>,
connection_status: Arc<Mutex<bool>>,
}
impl MQTTTarget {
pub async fn new(id: &str, args: MQTTArgs, logger: Logger) -> Result<Self, StoreError> {
// 创建取消通道
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::<Event>::new(queue_dir, args.queue_limit, Some(".event")));
queue_store.open().await?;
store = Some(queue_store.clone() as Arc<dyn Store<Event>>);
// 设置事件流
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<Arc<Mutex<AsyncClient>>>,
logger: Logger,
cancel_tx: mpsc::Sender<()>,
connection_status: Arc<Mutex<bool>>,
}
#[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<bool> {
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<bool> {
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<T>(&self) -> Option<Arc<dyn Store<T>>>
where
T: Serialize + DeserializeOwned + Send + Sync + 'static,
{
if let Some(store) = &self.store {
// 类型检查确保 T 是 Event 类型
if std::any::TypeId::of::<T>() == std::any::TypeId::of::<Event>() {
// 安全：我们已经检查类型 ID 匹配
let store_ptr = Arc::as_ptr(store);
let store_t = unsafe { Arc::from_raw(store_ptr as *const dyn Store<T>) };
// 增加引用计数，避免释放原始指针
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
}
}
crates/event/src/target/webhook.rs
+328
View File
@@ -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<Arc<dyn Store<Event>>>,
logger: Logger,
cancel_tx: mpsc::Sender<()>,
addr: String, // 完整地址，包含 IP/DNS 和端口号
}
impl WebhookTarget {
pub async fn new(id: &str, args: WebhookArgs, logger: Logger) -> Result<Self, StoreError> {
// 创建取消通道
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::<Event>::new(queue_dir, args.queue_limit, Some(".event")));
queue_store.open().await?;
store = Some(queue_store.clone() as Arc<dyn Store<Event>>);
// 设置事件流
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<bool> {
// 尝试连接到目标地址
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<bool> {
// 尝试连接到目标地址
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<T>(&self) -> Option<Arc<dyn Store<T>>>
where
T: Serialize + DeserializeOwned + Send + Sync + 'static,
{
if let Some(store) = &self.store {
// 注意：这里假设 T 是 Event 类型，需要类型转换（如果不是，将返回 None）
if std::any::TypeId::of::<T>() == std::any::TypeId::of::<Event>() {
// 安全：因为我们检查了类型 ID
let store_ptr = Arc::as_ptr(store);
let store_t = unsafe { Arc::from_raw(store_ptr as *const dyn Store<T>) };
// 增加引用计数，避免释放原始指针
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
}
}
crates/notify/src/adapter/mod.rs
+2 -2
View File
@@ -85,7 +85,7 @@ pub trait ChannelAdapter: Send + Sync + 'static {
}
/// Creates channel adapters based on the provided configuration.
pub fn create_adapters(configs: Vec<AdapterConfig>) -> Result<Vec<Arc<dyn ChannelAdapter>>, Error> {
pub async fn create_adapters(configs: Vec<AdapterConfig>) -> Result<Vec<Arc<dyn ChannelAdapter>>, Error> {
let mut adapters: Vec<Arc<dyn ChannelAdapter>> = Vec::new();
for config in configs {
@@ -93,7 +93,7 @@ pub fn create_adapters(configs: Vec<AdapterConfig>) -> Result<Vec<Arc<dyn Channe
#[cfg(feature = "webhook")]
AdapterConfig::Webhook(webhook_config) => {
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) => {
crates/notify/src/adapter/webhook.rs
+29 -38
View File
@@ -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<Arc<QueueStore<Event>>>,
/// 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(())
}
crates/notify/src/lib.rs
+4 -1
View File
@@ -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;
crates/notify/src/store/manager.rs
+1 -1
View File
@@ -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);
crates/notify/src/store/mod.rs
+12 -7
View File
@@ -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<Key>, mut done_ch: mpsc::Receiver<()>, logger: Logger) {
pub async fn send_items(
target: Arc<dyn Target>,
mut key_ch: mpsc::Receiver<Key>,
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<dyn Target>,
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<Key>, 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<Key>, mu
}
// 流式传输项目
pub async fn stream_items<T>(store: Arc<dyn Store<T>>, target: &dyn Target, done_ch: mpsc::Receiver<()>, logger: Logger)
pub async fn stream_items<T>(store: Arc<dyn Store<T>>, target: Arc<dyn Target>, done_ch: mpsc::Receiver<()>, logger: Logger)
where
T: Serialize + DeserializeOwned + Send + Sync + 'static,
{