From 8555d48ca29966c4dd02bac5f7a6b8dc48a23cd4 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 4 Jun 2025 23:46:14 +0800 Subject: [PATCH] fix --- crates/event/src/adapter/kafka.rs | 7 +- crates/event/src/adapter/mod.rs | 27 +- crates/event/src/adapter/mqtt.rs | 5 +- crates/event/src/adapter/webhook.rs | 18 +- crates/event/src/bus.rs | 92 ------ crates/event/src/config.rs | 449 ---------------------------- crates/event/src/config/adapter.rs | 42 +++ crates/event/src/config/kafka.rs | 44 +++ crates/event/src/config/mod.rs | 38 +++ crates/event/src/config/mqtt.rs | 46 +++ crates/event/src/config/notifier.rs | 73 +++++ crates/event/src/config/webhook.rs | 88 ++++++ crates/event/src/event_notifier.rs | 142 +++++++++ crates/event/src/event_system.rs | 81 +++++ crates/event/src/lib.rs | 18 +- crates/event/src/store/event.rs | 60 ---- crates/event/src/store/manager.rs | 2 +- crates/event/src/store/mod.rs | 1 - crates/event/src/store/queue.rs | 79 ++++- ecstore/src/config/com.rs | 58 ++-- ecstore/src/config/mod.rs | 8 + rustfs/src/event.rs | 1 + rustfs/src/main.rs | 2 +- 23 files changed, 717 insertions(+), 664 deletions(-) delete mode 100644 crates/event/src/bus.rs delete mode 100644 crates/event/src/config.rs create mode 100644 crates/event/src/config/adapter.rs create mode 100644 crates/event/src/config/kafka.rs create mode 100644 crates/event/src/config/mod.rs create mode 100644 crates/event/src/config/mqtt.rs create mode 100644 crates/event/src/config/notifier.rs create mode 100644 crates/event/src/config/webhook.rs create mode 100644 crates/event/src/event_notifier.rs create mode 100644 crates/event/src/event_system.rs delete mode 100644 crates/event/src/store/event.rs diff --git a/crates/event/src/adapter/kafka.rs b/crates/event/src/adapter/kafka.rs index c480ef23f..a774331bc 100644 --- a/crates/event/src/adapter/kafka.rs +++ b/crates/event/src/adapter/kafka.rs @@ -1,8 +1,7 @@ -use crate::config::{default_queue_limit, STORE_PREFIX}; -use crate::KafkaConfig; +use crate::config::kafka::KafkaConfig; +use crate::config::{default_queue_limit, DEFAULT_RETRY_INTERVAL, STORE_PREFIX}; use crate::{ChannelAdapter, ChannelAdapterType}; -use crate::{Error, QueueStore}; -use crate::{Event, DEFAULT_RETRY_INTERVAL}; +use crate::{Error, Event, QueueStore}; use async_trait::async_trait; use rdkafka::error::KafkaError; use rdkafka::producer::{FutureProducer, FutureRecord}; diff --git a/crates/event/src/adapter/mod.rs b/crates/event/src/adapter/mod.rs index a2d6471c2..89c01a422 100644 --- a/crates/event/src/adapter/mod.rs +++ b/crates/event/src/adapter/mod.rs @@ -1,6 +1,5 @@ -use crate::AdapterConfig; -use crate::Error; -use crate::Event; +use crate::config::adapter::AdapterConfig; +use crate::{Error, Event}; use async_trait::async_trait; use std::sync::Arc; @@ -11,6 +10,26 @@ pub(crate) mod mqtt; #[cfg(feature = "webhook")] pub(crate) mod webhook; +#[allow(dead_code)] +const NOTIFY_KAFKA_SUB_SYS: &str = "notify_kafka"; +#[allow(dead_code)] +const NOTIFY_MQTT_SUB_SYS: &str = "notify_mqtt"; +#[allow(dead_code)] +const NOTIFY_MY_SQL_SUB_SYS: &str = "notify_mysql"; +#[allow(dead_code)] +const NOTIFY_NATS_SUB_SYS: &str = "notify_nats"; +#[allow(dead_code)] +const NOTIFY_NSQ_SUB_SYS: &str = "notify_nsq"; +#[allow(dead_code)] +const NOTIFY_ES_SUB_SYS: &str = "notify_elasticsearch"; +#[allow(dead_code)] +const NOTIFY_AMQP_SUB_SYS: &str = "notify_amqp"; +#[allow(dead_code)] +const NOTIFY_POSTGRES_SUB_SYS: &str = "notify_postgres"; +#[allow(dead_code)] +const NOTIFY_REDIS_SUB_SYS: &str = "notify_redis"; +const NOTIFY_WEBHOOK_SUB_SYS: &str = "notify_webhook"; + /// The `ChannelAdapterType` enum represents the different types of channel adapters. /// /// It is used to identify the type of adapter being used in the system. @@ -68,7 +87,7 @@ pub trait ChannelAdapter: Send + Sync + 'static { } /// Creates channel adapters based on the provided configuration. -pub fn create_adapters(configs: &[AdapterConfig]) -> Result>, Error> { +pub fn create_adapters(configs: Vec) -> Result>, Error> { let mut adapters: Vec> = Vec::new(); for config in configs { diff --git a/crates/event/src/adapter/mqtt.rs b/crates/event/src/adapter/mqtt.rs index ffd3b4583..e60b17275 100644 --- a/crates/event/src/adapter/mqtt.rs +++ b/crates/event/src/adapter/mqtt.rs @@ -1,7 +1,6 @@ -use crate::Error; -use crate::Event; -use crate::MqttConfig; +use crate::config::mqtt::MqttConfig; use crate::{ChannelAdapter, ChannelAdapterType}; +use crate::{Error, Event}; use async_trait::async_trait; use rumqttc::{AsyncClient, MqttOptions, QoS}; use std::time::Duration; diff --git a/crates/event/src/adapter/webhook.rs b/crates/event/src/adapter/webhook.rs index 22884646d..1a8e171de 100644 --- a/crates/event/src/adapter/webhook.rs +++ b/crates/event/src/adapter/webhook.rs @@ -1,6 +1,6 @@ +use crate::config::webhook::WebhookConfig; use crate::config::STORE_PREFIX; use crate::store::queue::Store; -use crate::WebhookConfig; use crate::{ChannelAdapter, ChannelAdapterType}; use crate::{Error, QueueStore}; use crate::{Event, DEFAULT_RETRY_INTERVAL}; @@ -14,6 +14,22 @@ use std::time::Duration; use tokio::time::sleep; use ChannelAdapterType::Webhook; +// Webhook constants +pub const WEBHOOK_ENDPOINT: &str = "endpoint"; +pub const WEBHOOK_AUTH_TOKEN: &str = "auth_token"; +pub const WEBHOOK_QUEUE_DIR: &str = "queue_dir"; +pub const WEBHOOK_QUEUE_LIMIT: &str = "queue_limit"; +pub const WEBHOOK_CLIENT_CERT: &str = "client_cert"; +pub const WEBHOOK_CLIENT_KEY: &str = "client_key"; + +pub const ENV_WEBHOOK_ENABLE: &str = "RUSTFS_NOTIFY_WEBHOOK_ENABLE"; +pub const ENV_WEBHOOK_ENDPOINT: &str = "RUSTFS_NOTIFY_WEBHOOK_ENDPOINT"; +pub const ENV_WEBHOOK_AUTH_TOKEN: &str = "RUSTFS_NOTIFY_WEBHOOK_AUTH_TOKEN"; +pub const ENV_WEBHOOK_QUEUE_DIR: &str = "RUSTFS_NOTIFY_WEBHOOK_QUEUE_DIR"; +pub const ENV_WEBHOOK_QUEUE_LIMIT: &str = "RUSTFS_NOTIFY_WEBHOOK_QUEUE_LIMIT"; +pub const ENV_WEBHOOK_CLIENT_CERT: &str = "RUSTFS_NOTIFY_WEBHOOK_CLIENT_CERT"; +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 diff --git a/crates/event/src/bus.rs b/crates/event/src/bus.rs deleted file mode 100644 index c68a6a1d0..000000000 --- a/crates/event/src/bus.rs +++ /dev/null @@ -1,92 +0,0 @@ -use crate::ChannelAdapter; -use crate::Error; -use crate::EventStore; -use crate::{Event, Log}; -use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; -use tokio::sync::mpsc; -use tokio::time::Duration; -use tokio_util::sync::CancellationToken; -use tracing::instrument; - -/// Handles incoming events from the producer. -/// -/// This function is responsible for receiving events from the producer and sending them to the appropriate adapters. -/// It also handles the shutdown process and saves any pending logs to the event store. -#[instrument(skip_all)] -pub async fn event_bus( - mut rx: mpsc::Receiver, - adapters: Vec>, - store: Arc, - shutdown: CancellationToken, - shutdown_complete: Option>, -) -> Result<(), Error> { - let mut unprocessed_events = Vec::new(); - loop { - tokio::select! { - Some(event) = rx.recv() => { - let mut send_tasks = Vec::new(); - for adapter in &adapters { - if event.channels.contains(&adapter.name()) { - let adapter = adapter.clone(); - let event = event.clone(); - send_tasks.push(tokio::spawn(async move { - if let Err(e) = adapter.send(&event).await { - tracing::error!("Failed to send event to {}: {}", adapter.name(), e); - Err(e) - } else { - Ok(()) - } - })); - } - } - for task in send_tasks { - if task.await?.is_err() { - // If sending fails, add the event to the unprocessed list - let failed_event = event.clone(); - unprocessed_events.push(failed_event); - } - } - } - _ = shutdown.cancelled() => { - tracing::info!("Shutting down event bus, saving pending logs..."); - // Check if there are still unprocessed messages in the channel - while let Ok(Some(event)) = tokio::time::timeout( - Duration::from_millis(100), - rx.recv() - ).await { - unprocessed_events.push(event); - } - - // save only if there are unprocessed events - if !unprocessed_events.is_empty() { - tracing::info!("Save {} unhandled events", unprocessed_events.len()); - // create and save logging - let shutdown_log = Log { - event_name: crate::event::Name::Everything, - key: format!("shutdown_{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()), - records: unprocessed_events, - }; - - store.save_logs(&[shutdown_log]).await?; - } else { - tracing::info!("no unhandled events need to be saved"); - } - tracing::debug!("shutdown_complete is Some: {}", shutdown_complete.is_some()); - - if let Some(complete_sender) = shutdown_complete { - // send a completion signal - let result = complete_sender.send(()); - match result { - Ok(_) => tracing::info!("Event bus shutdown signal sent"), - Err(e) => tracing::error!("Failed to send event bus shutdown signal: {:?}", e), - } - tracing::info!("Shutting down event bus"); - } - tracing::info!("Event bus shutdown complete"); - break; - } - } - } - Ok(()) -} diff --git a/crates/event/src/config.rs b/crates/event/src/config.rs deleted file mode 100644 index 89156b805..000000000 --- a/crates/event/src/config.rs +++ /dev/null @@ -1,449 +0,0 @@ -use config::{Config, File, FileFormat}; -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::env; -use std::path::Path; -use tracing::info; - -/// The default configuration file name -const DEFAULT_CONFIG_FILE: &str = "event"; - -/// The prefix for the configuration file -pub const STORE_PREFIX: &str = "rustfs"; - -/// The default retry interval for the webhook adapter -pub const DEFAULT_RETRY_INTERVAL: u64 = 3; - -/// The default maximum retry count for the webhook adapter -pub const DEFAULT_MAX_RETRIES: u32 = 3; - -/// Add a common field for the adapter configuration -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct AdapterCommon { - /// Adapter identifier for unique identification - pub identifier: String, - /// Adapter description information - pub comment: String, - /// Whether to enable this adapter - #[serde(default)] - pub enable: bool, - #[serde(default = "default_queue_dir")] - pub queue_dir: String, - #[serde(default = "default_queue_limit")] - pub queue_limit: u64, -} - -impl Default for AdapterCommon { - fn default() -> Self { - Self { - identifier: String::new(), - comment: String::new(), - enable: false, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - } - } -} - -/// Configuration for the webhook adapter. -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct WebhookConfig { - #[serde(flatten)] - pub common: AdapterCommon, - pub endpoint: String, - pub auth_token: Option, - pub custom_headers: Option>, - pub max_retries: u32, - pub retry_interval: Option, - pub timeout: Option, - #[serde(default)] - pub client_cert: Option, - #[serde(default)] - pub client_key: Option, -} - -impl WebhookConfig { - /// validate the configuration for the webhook adapter - /// - /// # Returns - /// - /// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid. - /// - /// # Example - /// - /// ``` - /// use rustfs_event::WebhookConfig; - /// use rustfs_event::AdapterCommon; - /// use rustfs_event::DEFAULT_RETRY_INTERVAL; - /// - /// let config = WebhookConfig { - /// common: AdapterCommon::default(), - /// endpoint: "https://example.com/webhook".to_string(), - /// auth_token: Some("my_token".to_string()), - /// custom_headers: None, - /// max_retries: 3, - /// retry_interval: Some(DEFAULT_RETRY_INTERVAL), - /// timeout: Some(5000), - /// client_cert: None, - /// client_key: None, - /// }; - /// - /// assert!(config.validate().is_ok()); - pub fn validate(&self) -> Result<(), String> { - // If not enabled, the other fields are not validated - if !self.common.enable { - return Ok(()); - } - - // verify that endpoint cannot be empty - if self.endpoint.trim().is_empty() { - return Err("Webhook endpoint cannot be empty".to_string()); - } - - // verification timeout must be reasonable - if self.timeout.is_some() { - match self.timeout { - Some(timeout) if timeout > 0 => { - info!("Webhook timeout is set to {}", timeout); - } - _ => return Err("Webhook timeout must be greater than 0".to_string()), - } - } - - // Verify that the maximum number of retry is reasonable - if self.max_retries > 10 { - return Err("Maximum retry count cannot exceed 10".to_string()); - } - - // Verify the queue directory path - if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() { - return Err("Queue directory path should be absolute".to_string()); - } - - // The authentication certificate and key must appear in pairs - if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some()) - { - return Err("Certificate and key must be specified as a pair".to_string()); - } - - Ok(()) - } - - /// Create a new webhook configuration - pub fn new(identifier: impl Into, endpoint: impl Into) -> Self { - Self { - common: AdapterCommon { - identifier: identifier.into(), - comment: String::new(), - enable: true, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - }, - endpoint: endpoint.into(), - ..Default::default() - } - } -} - -/// Configuration for the Kafka adapter. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct KafkaConfig { - #[serde(flatten)] - pub common: AdapterCommon, - pub brokers: String, - pub topic: String, - pub max_retries: u32, - pub timeout: u64, -} - -impl Default for KafkaConfig { - fn default() -> Self { - Self { - common: AdapterCommon::default(), - brokers: String::new(), - topic: String::new(), - max_retries: 3, - timeout: 5000, - } - } -} - -impl KafkaConfig { - /// Create a new Kafka configuration - pub fn new(identifier: impl Into, brokers: impl Into, topic: impl Into) -> Self { - Self { - common: AdapterCommon { - identifier: identifier.into(), - comment: String::new(), - enable: true, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - }, - brokers: brokers.into(), - topic: topic.into(), - ..Default::default() - } - } -} - -/// Configuration for the MQTT adapter. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct MqttConfig { - #[serde(flatten)] - pub common: AdapterCommon, - pub broker: String, - pub port: u16, - pub client_id: String, - pub topic: String, - pub max_retries: u32, -} - -impl Default for MqttConfig { - fn default() -> Self { - Self { - common: AdapterCommon::default(), - broker: String::new(), - port: 1883, - client_id: String::new(), - topic: String::new(), - max_retries: 3, - } - } -} - -impl MqttConfig { - /// Create a new MQTT configuration - pub fn new(identifier: impl Into, broker: impl Into, topic: impl Into) -> Self { - Self { - common: AdapterCommon { - identifier: identifier.into(), - comment: String::new(), - enable: true, - queue_dir: default_queue_dir(), - queue_limit: default_queue_limit(), - }, - broker: broker.into(), - topic: topic.into(), - ..Default::default() - } - } -} - -/// Configuration for the adapter. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(tag = "type")] -pub enum AdapterConfig { - Webhook(WebhookConfig), - Kafka(KafkaConfig), - Mqtt(MqttConfig), -} - -/// Configuration for the notification system. -/// -/// This struct contains the configuration for the notification system, including -/// the storage path, channel capacity, and a list of adapters. -/// -/// # Fields -/// -/// - `store_path`: The path to the storage directory. -/// - `channel_capacity`: The capacity of the notification channel. -/// - `adapters`: A list of adapters to be used for notifications. -/// -/// # Example -/// -/// ``` -/// use rustfs_event::NotifierConfig; -/// -/// let config = NotifierConfig::new(); -/// ``` -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct NotifierConfig { - #[serde(default = "default_queue_dir")] - pub store_path: String, - #[serde(default = "default_queue_limit")] - pub channel_capacity: u64, - pub adapters: Vec, -} - -impl Default for NotifierConfig { - fn default() -> Self { - Self { - store_path: default_queue_dir(), - channel_capacity: default_queue_limit(), - adapters: Vec::new(), - } - } -} - -impl NotifierConfig { - /// create a new configuration with default values - pub fn new() -> Self { - Self::default() - } - - /// Loading the configuration file - /// Supports TOML, YAML and .env formats, read in order by priority - /// - /// # Parameters - /// - `config_dir`: Configuration file path - /// - /// # Returns - /// Configuration information - /// - /// # Example - /// ``` - /// use rustfs_event::NotifierConfig; - /// - /// let config = NotifierConfig::event_load_config(None); - /// ``` - pub fn event_load_config(config_dir: Option) -> NotifierConfig { - let config_dir = if let Some(path) = config_dir { - // If a path is provided, check if it's empty - if path.is_empty() { - // If empty, use the default config file name - DEFAULT_CONFIG_FILE.to_string() - } else { - // Use the provided path - let path = Path::new(&path); - if path.extension().is_some() { - // If path has extension, use it as is (extension will be added by Config::builder) - path.with_extension("").to_string_lossy().into_owned() - } else { - // If path is a directory, append the default config file name - path.to_string_lossy().into_owned() - } - } - } else { - // If no path provided, use current directory + default config file - match env::current_dir() { - Ok(dir) => dir.join(DEFAULT_CONFIG_FILE).to_string_lossy().into_owned(), - Err(_) => { - eprintln!("Warning: Failed to get current directory, using default config file"); - DEFAULT_CONFIG_FILE.to_string() - } - } - }; - - // Log using proper logging instead of println when possible - println!("Using config file base: {}", config_dir); - - let app_config = Config::builder() - .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false)) - .add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false)) - .build() - .unwrap_or_default(); - match app_config.try_deserialize::() { - Ok(app_config) => { - println!("Parsed AppConfig: {:?} \n", app_config); - app_config - } - Err(e) => { - println!("Failed to deserialize config: {}", e); - NotifierConfig::default() - } - } - } - - /// unmarshal the configuration from a byte array - pub fn unmarshal(data: &[u8]) -> common::error::Result { - let m: NotifierConfig = serde_json::from_slice(data)?; - Ok(m) - } - - /// marshal the configuration to a byte array - pub fn marshal(&self) -> common::error::Result> { - let data = serde_json::to_vec(&self)?; - Ok(data) - } - - /// merge the configuration with default values - pub fn merge(&self) -> NotifierConfig { - self.clone() - } -} - -/// Provide temporary directories as default storage paths -fn default_queue_dir() -> String { - env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| { - tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string()); - env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string() - }) -} - -/// Provides the recommended default channel capacity for high concurrency systems -pub(crate) fn default_queue_limit() -> u64 { - env::var("EVENT_CHANNEL_CAPACITY") - .unwrap_or_else(|_| "10000".to_string()) - .parse() - .unwrap_or(10000) // Default to 10000 if parsing fails -} - -/// Event Notifier Configuration -/// This struct contains the configuration for the event notifier system, -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct EventNotifierConfig { - /// A collection of webhook configurations, with the key being a unique identifier - #[serde(default)] - pub webhook: HashMap, - /// A collection of Kafka configurations, with the key being a unique identifier - #[serde(default)] - pub kafka: HashMap, - ///MQTT configuration collection, with the key being a unique identifier - #[serde(default)] - pub mqtt: HashMap, -} - -impl EventNotifierConfig { - /// Create a new default configuration - pub fn new() -> Self { - Self::default() - } - - /// Load the configuration from the file - pub fn event_load_config(_config_dir: Option) -> EventNotifierConfig { - // The existing implementation remains the same, but returns EventNotifierConfig - // ... - - Self::default() - } - - /// Deserialization configuration - pub fn unmarshal(data: &[u8]) -> common::error::Result { - let m: EventNotifierConfig = serde_json::from_slice(data)?; - Ok(m) - } - - /// Serialization configuration - pub fn marshal(&self) -> common::error::Result> { - let data = serde_json::to_vec(&self)?; - Ok(data) - } - - /// Convert this configuration to a list of adapter configurations - pub fn to_adapter_configs(&self) -> Vec { - let mut adapters = Vec::new(); - - // Add all enabled webhook configurations - for webhook in self.webhook.values() { - if webhook.common.enable { - adapters.push(AdapterConfig::Webhook(webhook.clone())); - } - } - - // Add all enabled Kafka configurations - for kafka in self.kafka.values() { - if kafka.common.enable { - adapters.push(AdapterConfig::Kafka(kafka.clone())); - } - } - - // Add all enabled MQTT configurations - for mqtt in self.mqtt.values() { - if mqtt.common.enable { - adapters.push(AdapterConfig::Mqtt(mqtt.clone())); - } - } - - adapters - } -} diff --git a/crates/event/src/config/adapter.rs b/crates/event/src/config/adapter.rs new file mode 100644 index 000000000..19dc0978a --- /dev/null +++ b/crates/event/src/config/adapter.rs @@ -0,0 +1,42 @@ +use crate::config::kafka::KafkaConfig; +use crate::config::mqtt::MqttConfig; +use crate::config::webhook::WebhookConfig; +use crate::config::{default_queue_dir, default_queue_limit}; +use serde::{Deserialize, Serialize}; + +/// Add a common field for the adapter configuration +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AdapterCommon { + /// Adapter identifier for unique identification + pub identifier: String, + /// Adapter description information + pub comment: String, + /// Whether to enable this adapter + #[serde(default)] + pub enable: bool, + #[serde(default = "default_queue_dir")] + pub queue_dir: String, + #[serde(default = "default_queue_limit")] + pub queue_limit: u64, +} + +impl Default for AdapterCommon { + fn default() -> Self { + Self { + identifier: String::new(), + comment: String::new(), + enable: false, + queue_dir: default_queue_dir(), + queue_limit: default_queue_limit(), + } + } +} + +/// Configuration for the adapter. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type")] +pub enum AdapterConfig { + Webhook(WebhookConfig), + Kafka(KafkaConfig), + Mqtt(MqttConfig), +} diff --git a/crates/event/src/config/kafka.rs b/crates/event/src/config/kafka.rs new file mode 100644 index 000000000..84a789e5c --- /dev/null +++ b/crates/event/src/config/kafka.rs @@ -0,0 +1,44 @@ +use crate::config::adapter::AdapterCommon; +use crate::config::{default_queue_dir, default_queue_limit}; +use serde::{Deserialize, Serialize}; + +/// Configuration for the Kafka adapter. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct KafkaConfig { + #[serde(flatten)] + pub common: AdapterCommon, + pub brokers: String, + pub topic: String, + pub max_retries: u32, + pub timeout: u64, +} + +impl Default for KafkaConfig { + fn default() -> Self { + Self { + common: AdapterCommon::default(), + brokers: String::new(), + topic: String::new(), + max_retries: 3, + timeout: 5000, + } + } +} + +impl KafkaConfig { + /// Create a new Kafka configuration + pub fn new(identifier: impl Into, brokers: impl Into, topic: impl Into) -> Self { + Self { + common: AdapterCommon { + identifier: identifier.into(), + comment: String::new(), + enable: true, + queue_dir: default_queue_dir(), + queue_limit: default_queue_limit(), + }, + brokers: brokers.into(), + topic: topic.into(), + ..Default::default() + } + } +} diff --git a/crates/event/src/config/mod.rs b/crates/event/src/config/mod.rs new file mode 100644 index 000000000..fa69e5a48 --- /dev/null +++ b/crates/event/src/config/mod.rs @@ -0,0 +1,38 @@ +use std::env; + +pub mod adapter; +pub mod kafka; +pub mod mqtt; +pub mod notifier; +pub mod webhook; + +/// The default configuration file name +const DEFAULT_CONFIG_FILE: &str = "event"; + +/// The prefix for the configuration file +pub const STORE_PREFIX: &str = "rustfs"; + +/// The default retry interval for the webhook adapter +pub const DEFAULT_RETRY_INTERVAL: u64 = 3; + +/// The default maximum retry count for the webhook adapter +pub const DEFAULT_MAX_RETRIES: u32 = 3; + +/// The default notification queue limit +pub const DEFAULT_NOTIFY_QUEUE_LIMIT: u64 = 10000; + +/// Provide temporary directories as default storage paths +pub(crate) fn default_queue_dir() -> String { + env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| { + tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string()); + env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string() + }) +} + +/// Provides the recommended default channel capacity for high concurrency systems +pub(crate) fn default_queue_limit() -> u64 { + env::var("EVENT_CHANNEL_CAPACITY") + .unwrap_or_else(|_| DEFAULT_NOTIFY_QUEUE_LIMIT.to_string()) + .parse() + .unwrap_or(DEFAULT_NOTIFY_QUEUE_LIMIT) // Default to 10000 if parsing fails +} diff --git a/crates/event/src/config/mqtt.rs b/crates/event/src/config/mqtt.rs new file mode 100644 index 000000000..5090fcc91 --- /dev/null +++ b/crates/event/src/config/mqtt.rs @@ -0,0 +1,46 @@ +use crate::config::adapter::AdapterCommon; +use crate::config::{default_queue_dir, default_queue_limit}; +use serde::{Deserialize, Serialize}; + +/// Configuration for the MQTT adapter. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MqttConfig { + #[serde(flatten)] + pub common: AdapterCommon, + pub broker: String, + pub port: u16, + pub client_id: String, + pub topic: String, + pub max_retries: u32, +} + +impl Default for MqttConfig { + fn default() -> Self { + Self { + common: AdapterCommon::default(), + broker: String::new(), + port: 1883, + client_id: String::new(), + topic: String::new(), + max_retries: 3, + } + } +} + +impl MqttConfig { + /// Create a new MQTT configuration + pub fn new(identifier: impl Into, broker: impl Into, topic: impl Into) -> Self { + Self { + common: AdapterCommon { + identifier: identifier.into(), + comment: String::new(), + enable: true, + queue_dir: default_queue_dir(), + queue_limit: default_queue_limit(), + }, + broker: broker.into(), + topic: topic.into(), + ..Default::default() + } + } +} diff --git a/crates/event/src/config/notifier.rs b/crates/event/src/config/notifier.rs new file mode 100644 index 000000000..5d0658c73 --- /dev/null +++ b/crates/event/src/config/notifier.rs @@ -0,0 +1,73 @@ +use crate::config::{adapter::AdapterConfig, kafka::KafkaConfig, mqtt::MqttConfig, webhook::WebhookConfig}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; + +/// Event Notifier Configuration +/// This struct contains the configuration for the event notifier system, +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct EventNotifierConfig { + /// A collection of webhook configurations, with the key being a unique identifier + #[serde(default)] + pub webhook: HashMap, + /// A collection of Kafka configurations, with the key being a unique identifier + #[serde(default)] + pub kafka: HashMap, + ///MQTT configuration collection, with the key being a unique identifier + #[serde(default)] + pub mqtt: HashMap, +} + +impl EventNotifierConfig { + /// Create a new default configuration + pub fn new() -> Self { + Self::default() + } + + /// Load the configuration from the file + pub fn event_load_config(_config_dir: Option) -> EventNotifierConfig { + // The existing implementation remains the same, but returns EventNotifierConfig + // ... + + Self::default() + } + + /// Deserialization configuration + pub fn unmarshal(data: &[u8]) -> common::error::Result { + let m: EventNotifierConfig = serde_json::from_slice(data)?; + Ok(m) + } + + /// Serialization configuration + pub fn marshal(&self) -> common::error::Result> { + let data = serde_json::to_vec(&self)?; + Ok(data) + } + + /// Convert this configuration to a list of adapter configurations + pub fn to_adapter_configs(&self) -> Vec { + let mut adapters = Vec::new(); + + // Add all enabled webhook configurations + for webhook in self.webhook.values() { + if webhook.common.enable { + adapters.push(AdapterConfig::Webhook(webhook.clone())); + } + } + + // Add all enabled Kafka configurations + for kafka in self.kafka.values() { + if kafka.common.enable { + adapters.push(AdapterConfig::Kafka(kafka.clone())); + } + } + + // Add all enabled MQTT configurations + for mqtt in self.mqtt.values() { + if mqtt.common.enable { + adapters.push(AdapterConfig::Mqtt(mqtt.clone())); + } + } + + adapters + } +} diff --git a/crates/event/src/config/webhook.rs b/crates/event/src/config/webhook.rs new file mode 100644 index 000000000..79744110d --- /dev/null +++ b/crates/event/src/config/webhook.rs @@ -0,0 +1,88 @@ +use crate::config::adapter::AdapterCommon; +use crate::config::{default_queue_dir, default_queue_limit}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::path::Path; +use tracing::info; + +/// Configuration for the webhook adapter. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct WebhookConfig { + #[serde(flatten)] + pub common: AdapterCommon, + pub endpoint: String, + pub auth_token: Option, + pub custom_headers: Option>, + pub max_retries: u32, + pub retry_interval: Option, + pub timeout: Option, + #[serde(default)] + pub client_cert: Option, + #[serde(default)] + pub client_key: Option, +} + +impl WebhookConfig { + /// validate the configuration for the webhook adapter + /// + /// # Returns + /// + /// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid. + /// + /// # Errors + /// - Returns an error if the configuration is invalid, such as empty endpoint, unreasonable timeout, or mismatched certificate and key. + pub fn validate(&self) -> Result<(), String> { + // If not enabled, the other fields are not validated + if !self.common.enable { + return Ok(()); + } + + // verify that endpoint cannot be empty + if self.endpoint.trim().is_empty() { + return Err("Webhook endpoint cannot be empty".to_string()); + } + + // verification timeout must be reasonable + if self.timeout.is_some() { + match self.timeout { + Some(timeout) if timeout > 0 => { + info!("Webhook timeout is set to {}", timeout); + } + _ => return Err("Webhook timeout must be greater than 0".to_string()), + } + } + + // Verify that the maximum number of retry is reasonable + if self.max_retries > 10 { + return Err("Maximum retry count cannot exceed 10".to_string()); + } + + // Verify the queue directory path + if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() { + return Err("Queue directory path should be absolute".to_string()); + } + + // The authentication certificate and key must appear in pairs + if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some()) + { + return Err("Certificate and key must be specified as a pair".to_string()); + } + + Ok(()) + } + + /// Create a new webhook configuration + pub fn new(identifier: impl Into, endpoint: impl Into) -> Self { + Self { + common: AdapterCommon { + identifier: identifier.into(), + comment: String::new(), + enable: true, + queue_dir: default_queue_dir(), + queue_limit: default_queue_limit(), + }, + endpoint: endpoint.into(), + ..Default::default() + } + } +} diff --git a/crates/event/src/event_notifier.rs b/crates/event/src/event_notifier.rs new file mode 100644 index 000000000..c9c80d2e2 --- /dev/null +++ b/crates/event/src/event_notifier.rs @@ -0,0 +1,142 @@ +use crate::config::notifier::EventNotifierConfig; +use crate::Event; +use common::error::{Error, Result}; +use ecstore::store::ECStore; +use std::sync::Arc; +use tokio::sync::{broadcast, mpsc}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, instrument, warn}; +use tracing_subscriber::util::SubscriberInitExt; + +/// 事件通知器 +pub struct EventNotifier { + /// 事件发送通道 + sender: mpsc::Sender, + /// 接收器任务句柄 + task_handle: Option>, + /// 配置信息 + config: EventNotifierConfig, + /// 关闭标记 + shutdown: CancellationToken, + /// 关闭通知通道 + shutdown_complete_tx: Option>, +} + +impl EventNotifier { + /// 创建新的事件通知器 + #[instrument(skip_all)] + pub async fn new(store: Arc) -> Result { + let manager = crate::store::manager::EventManager::new(store); + + // 初始化配置 + let config = manager.init().await?; + + // 创建适配器 + let adapters = manager.create_adapters().await?; + info!("创建了 {} 个适配器", adapters.len()); + + // 创建关闭标记 + let shutdown = CancellationToken::new(); + let (shutdown_complete_tx, _) = broadcast::channel(1); + + // 创建事件通道 - 使用默认容量，因为每个适配器都有自己的队列 + // 这里使用较小的通道容量，因为事件会被快速分发到适配器 + let (sender, mut receiver) = mpsc::channel(100); + + let shutdown_clone = shutdown.clone(); + let shutdown_complete_tx_clone = shutdown_complete_tx.clone(); + let adapters_clone = adapters.clone(); + + // 启动事件处理任务 + let task_handle = tokio::spawn(async move { + debug!("事件处理任务启动"); + + loop { + tokio::select! { + Some(event) = receiver.recv() => { + debug!("收到事件：{}", event.id); + + // 分发到所有适配器 + for adapter in &adapters_clone { + let adapter_name = adapter.name(); + match adapter.send(&event).await { + Ok(_) => { + debug!("事件 {} 成功发送到适配器 {}", event.id, adapter_name); + } + Err(e) => { + error!("事件 {} 发送到适配器 {} 失败：{}", event.id, adapter_name, e); + } + } + } + } + + _ = shutdown_clone.cancelled() => { + info!("接收到关闭信号，事件处理任务停止"); + let _ = shutdown_complete_tx_clone.send(()); + break; + } + } + } + + debug!("事件处理任务已停止"); + }); + + Ok(Self { + sender, + task_handle: Some(task_handle), + config, + shutdown, + shutdown_complete_tx: Some(shutdown_complete_tx), + }) + } + + /// 关闭事件通知器 + pub async fn shutdown(&mut self) -> Result<()> { + info!("关闭事件通知器"); + self.shutdown.cancel(); + + if let Some(shutdown_tx) = self.shutdown_complete_tx.take() { + let mut rx = shutdown_tx.subscribe(); + + // 等待关闭完成信号或超时 + tokio::select! { + _ = rx.recv() => { + debug!("收到关闭完成信号"); + } + _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => { + warn!("关闭超时，强制终止"); + } + } + } + + if let Some(handle) = self.task_handle.take() { + handle.abort(); + match handle.await { + Ok(_) => debug!("事件处理任务已正常终止"), + Err(e) => { + if e.is_cancelled() { + debug!("事件处理任务已取消"); + } else { + error!("等待事件处理任务终止时出错：{}", e); + } + } + } + } + + info!("事件通知器已完全关闭"); + Ok(()) + } + + /// 发送事件 + pub async fn send(&self, event: Event) -> Result<()> { + self.sender + .send(event) + .await + .map_err(|e| Error::msg(format!("发送事件到通道失败：{}", e))) + } + + /// 获取当前配置 + pub fn config(&self) -> &EventNotifierConfig { + &self.config + } +} diff --git a/crates/event/src/event_system.rs b/crates/event/src/event_system.rs new file mode 100644 index 000000000..77fd6636f --- /dev/null +++ b/crates/event/src/event_system.rs @@ -0,0 +1,81 @@ +use crate::config::notifier::EventNotifierConfig; +use crate::event_notifier::EventNotifier; +use common::error::Result; +use ecstore::store::ECStore; +use once_cell::sync::OnceCell; +use std::sync::{Arc, Mutex}; +use tracing::{debug, error, info}; + +/// 全局事件系统 +pub struct EventSystem { + /// 事件通知器 + notifier: Mutex>, +} + +impl EventSystem { + /// 创建一个新的事件系统 + pub fn new() -> Self { + Self { + notifier: Mutex::new(None), + } + } + + /// 初始化事件系统 + pub async fn init(&self, store: Arc) -> Result { + info!("初始化事件系统"); + let notifier = EventNotifier::new(store).await?; + let config = notifier.config().clone(); + + let mut guard = self + .notifier + .lock() + .map_err(|e| common::error::Error::msg(format!("获取锁失败：{}", e)))?; + + *guard = Some(notifier); + debug!("事件系统初始化完成"); + + Ok(config) + } + + /// 发送事件 + pub async fn send_event(&self, event: crate::Event) -> Result<()> { + let guard = self + .notifier + .lock() + .map_err(|e| common::error::Error::msg(format!("获取锁失败：{}", e)))?; + + if let Some(notifier) = &*guard { + notifier.send(event).await + } else { + error!("事件系统未初始化"); + Err(common::error::Error::msg("事件系统未初始化")) + } + } + + /// 关闭事件系统 + pub async fn shutdown(&self) -> Result<()> { + info!("关闭事件系统"); + let mut guard = self + .notifier + .lock() + .map_err(|e| common::error::Error::msg(format!("获取锁失败：{}", e)))?; + + if let Some(ref mut notifier) = *guard { + notifier.shutdown().await?; + *guard = None; + info!("事件系统已关闭"); + Ok(()) + } else { + debug!("事件系统已经关闭"); + Ok(()) + } + } +} + +/// 全局事件系统实例 +pub static GLOBAL_EVENT_SYS: OnceCell = OnceCell::new(); + +/// 初始化全局事件系统 +pub fn init_global_event_system() -> &'static EventSystem { + GLOBAL_EVENT_SYS.get_or_init(EventSystem::new) +} diff --git a/crates/event/src/lib.rs b/crates/event/src/lib.rs index 73411e44d..4b4759b4f 100644 --- a/crates/event/src/lib.rs +++ b/crates/event/src/lib.rs @@ -1,8 +1,9 @@ mod adapter; -mod bus; mod config; mod error; mod event; +mod event_notifier; +mod event_system; mod global; mod notifier; mod store; @@ -16,20 +17,19 @@ pub use adapter::mqtt::MqttAdapter; pub use adapter::webhook::WebhookAdapter; pub use adapter::ChannelAdapter; pub use adapter::ChannelAdapterType; -pub use bus::event_bus; -pub use config::AdapterCommon; -pub use config::EventNotifierConfig; +pub use config::adapter::AdapterCommon; +pub use config::adapter::AdapterConfig; #[cfg(all(feature = "kafka", target_os = "linux"))] -pub use config::KafkaConfig; +pub use config::kafka::KafkaConfig; #[cfg(feature = "mqtt")] -pub use config::MqttConfig; +pub use config::mqtt::MqttConfig; +pub use config::notifier::EventNotifierConfig; #[cfg(feature = "webhook")] -pub use config::WebhookConfig; -pub use config::{AdapterConfig, NotifierConfig, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL}; +pub use config::webhook::WebhookConfig; +pub use config::{DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL}; pub use error::Error; pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source}; pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; pub use notifier::NotifierSystem; -pub use store::event::EventStore; pub use store::queue::QueueStore; diff --git a/crates/event/src/store/event.rs b/crates/event/src/store/event.rs deleted file mode 100644 index 249116152..000000000 --- a/crates/event/src/store/event.rs +++ /dev/null @@ -1,60 +0,0 @@ -use crate::Error; -use crate::Log; -use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; -use tokio::fs::{create_dir_all, File, OpenOptions}; -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}; -use tokio::sync::RwLock; -use tracing::instrument; - -/// `EventStore` is a struct that manages the storage of event logs. -pub struct EventStore { - path: String, - lock: Arc>, -} - -impl EventStore { - pub async fn new(path: &str) -> Result { - create_dir_all(path).await?; - Ok(Self { - path: path.to_string(), - lock: Arc::new(RwLock::new(())), - }) - } - - #[instrument(skip(self))] - pub async fn save_logs(&self, logs: &[Log]) -> Result<(), Error> { - let _guard = self.lock.write().await; - let file_path = format!( - "{}/events_{}.jsonl", - self.path, - SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() - ); - let file = OpenOptions::new().create(true).append(true).open(&file_path).await?; - let mut writer = BufWriter::new(file); - for log in logs { - let line = serde_json::to_string(log)?; - writer.write_all(line.as_bytes()).await?; - writer.write_all(b"\n").await?; - } - writer.flush().await?; - tracing::info!("Saved logs to {} end", file_path); - Ok(()) - } - - pub async fn load_logs(&self) -> Result, Error> { - let _guard = self.lock.read().await; - let mut logs = Vec::new(); - let mut entries = tokio::fs::read_dir(&self.path).await?; - while let Some(entry) = entries.next_entry().await? { - let file = File::open(entry.path()).await?; - let reader = BufReader::new(file); - let mut lines = reader.lines(); - while let Some(line) = lines.next_line().await? { - let log: Log = serde_json::from_str(&line)?; - logs.push(log); - } - } - Ok(logs) - } -} diff --git a/crates/event/src/store/manager.rs b/crates/event/src/store/manager.rs index b8534e91e..06841e15e 100644 --- a/crates/event/src/store/manager.rs +++ b/crates/event/src/store/manager.rs @@ -162,7 +162,7 @@ impl EventManager { }; let adapter_configs = config.to_adapter_configs(); - match adapter::create_adapters(&adapter_configs) { + match adapter::create_adapters(adapter_configs) { Ok(adapters) => Ok(adapters), Err(err) => { tracing::error!("Failed to create adapters: {:?}", err); diff --git a/crates/event/src/store/mod.rs b/crates/event/src/store/mod.rs index 4081fed2d..081e4e2e0 100644 --- a/crates/event/src/store/mod.rs +++ b/crates/event/src/store/mod.rs @@ -1,3 +1,2 @@ -pub(crate) mod event; pub(crate) mod manager; pub(crate) mod queue; diff --git a/crates/event/src/store/queue.rs b/crates/event/src/store/queue.rs index f534b7400..9ffb66175 100644 --- a/crates/event/src/store/queue.rs +++ b/crates/event/src/store/queue.rs @@ -1,17 +1,18 @@ use common::error::{Error, Result}; +use ecstore::utils::path::dir; use serde::{de::DeserializeOwned, Serialize}; use snap::raw::{Decoder, Encoder}; use std::collections::HashMap; use std::io::Read; use std::marker::PhantomData; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::{Arc, RwLock}; use std::time::{SystemTime, UNIX_EPOCH}; use std::{fs, io}; use uuid::Uuid; /// Keys in storage -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct Key { /// Key name pub name: String, @@ -42,7 +43,7 @@ impl Key { key_str = format!("{}:{}", self.item_count, self.name); } - let compress_ext = if self.compress { ".snappy" } else { "" }; + let compress_ext = if self.compress { COMPRESS_EXT } else { "" }; format!("{}{}{}", key_str, self.extension, compress_ext) } } @@ -50,7 +51,7 @@ impl Key { /// Parse key from file name #[allow(clippy::redundant_closure)] pub fn parse_key(filename: &str) -> Key { - let compress = filename.ends_with(".snappy"); + let compress = filename.ends_with(COMPRESS_EXT); let filename = if compress { &filename[..filename.len() - 7] // 移除 ".snappy" } else { @@ -85,7 +86,7 @@ pub fn parse_key(filename: &str) -> Key { /// Store the characteristics of the project pub trait Store: Send + Sync where - T: Serialize + DeserializeOwned + Send + Sync, + T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, { /// Store a single item fn put(&self, item: T) -> Result; @@ -142,26 +143,88 @@ pub struct QueueStore { entries: Arc>>, /// Type tags _phantom: PhantomData, + /// Whether to compress + compress: bool, + /// Store name + name: String, } impl QueueStore where - T: Serialize + DeserializeOwned + Send + Sync, + T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, { /// Create a new queue store - pub fn new(directory: impl Into, limit: u64, ext: Option) -> Self { + pub fn new>(directory: P, name: String, limit: u64, ext: Option) -> Self { let limit = if limit == 0 { DEFAULT_LIMIT } else { limit }; let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string()); + let mut path = PathBuf::from(directory.as_ref()); + path.push(&name); + + // Create a directory (if it doesn't exist) + if !path.exists() { + if let Err(e) = fs::create_dir_all(&path) { + tracing::error!("创建存储目录失败 {}: {}", path.display(), e); + } + } Self { directory: directory.into(), + name, entry_limit: limit, file_ext: ext, + compress: true, // Default to compressing entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))), _phantom: PhantomData, } } + /// Set the file extension + pub fn with_file_ext(mut self, file_ext: &str) -> Self { + self.file_ext = file_ext.to_string(); + self + } + + /// Set whether to compress or not + pub fn with_compression(mut self, compress: bool) -> Self { + self.compress = compress; + self + } + + /// Get the file path + fn get_file_path(&self, key: &Key) -> PathBuf { + let mut filename = key.to_string(); + filename.push_str(if self.compress { COMPRESS_EXT } else { &self.file_ext }); + self.directory.join(filename) + } + + /// Serialize the project + fn serialize_item(&self, item: &T) -> Result> { + let data = serde_json::to_vec(item).map_err(|e| Error::msg(format!("Serialization failed: {}", e)))?; + + if self.compress { + let mut encoder = Encoder::new(); + Ok(encoder + .compress_vec(&data) + .map_err(|e| Error::msg(format!("Compression failed: {}", e)))?) + } else { + Ok(data) + } + } + + /// Deserialize the project + fn deserialize_item(&self, data: &[u8], is_compressed: bool) -> Result { + let data = if is_compressed { + let mut decoder = Decoder::new(); + decoder + .decompress_vec(data) + .map_err(|e| Error::msg(format!("Unzipping failed: {}", e)))? + } else { + data.to_vec() + }; + + serde_json::from_slice(&data).map_err(|e| Error::msg(format!("Deserialization failed: {}", e))) + } + /// Lists all files in the directory, sorted by modification time (oldest takes precedence.)） fn list_files(&self) -> Result> { let mut files = Vec::new(); @@ -264,7 +327,7 @@ where impl Store for QueueStore where - T: Serialize + DeserializeOwned + Send + Sync + Clone, + T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static, { fn open(&self) -> Result<()> { // Create a directory (if it doesn't exist) diff --git a/ecstore/src/config/com.rs b/ecstore/src/config/com.rs index 6ab51290e..59d22fa8e 100644 --- a/ecstore/src/config/com.rs +++ b/ecstore/src/config/com.rs @@ -114,22 +114,12 @@ async fn new_and_save_server_config(api: Arc) -> Result String { + format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE) +} + pub async fn read_config_without_migrate(api: Arc) -> Result { - let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); - let data = match read_config(api.clone(), config_file.as_str()).await { - Ok(res) => res, - Err(err) => { - return if is_err_config_not_found(&err) { - warn!("config not found, start to init"); - let cfg = new_and_save_server_config(api).await?; - warn!("config init done"); - Ok(cfg) - } else { - error!("read config err {:?}", &err); - Err(err) - } - } - }; + let data = handle_read_config(api.clone()).await?; read_server_config(api, data.as_slice()).await } @@ -137,21 +127,8 @@ pub async fn read_config_without_migrate(api: Arc) -> Result(api: Arc, data: &[u8]) -> Result { let cfg = { if data.is_empty() { - let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); - let cfg_data = match read_config(api.clone(), config_file.as_str()).await { - Ok(res) => res, - Err(err) => { - return if is_err_config_not_found(&err) { - warn!("config not found init start"); - let cfg = new_and_save_server_config(api).await?; - warn!("config not found init done"); - Ok(cfg) - } else { - error!("read config err {:?}", &err); - Err(err) - } - } - }; + let cfg_data = handle_read_config(api.clone()).await?; + // TODO: decrypt Config::unmarshal(cfg_data.as_slice())? @@ -163,10 +140,29 @@ async fn read_server_config(api: Arc, data: &[u8]) -> Result(api: Arc) -> Result> { + let config_file = get_config_file(); + match read_config(api.clone(), config_file.as_str()).await { + Ok(res) => Ok(res), + Err(err) => { + if is_err_config_not_found(&err) { + warn!("config not found, start to init"); + let cfg = new_and_save_server_config(api).await?; + warn!("config init done"); + // This returns the serialized data, keeping the interface consistent + cfg.marshal() + } else { + error!("read config err {:?}", &err); + Err(err) + } + } + } +} + async fn save_server_config(api: Arc, cfg: &Config) -> Result<()> { let data = cfg.marshal()?; - let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); + let config_file = get_config_file(); save_config(api, &config_file, data).await } diff --git a/ecstore/src/config/mod.rs b/ecstore/src/config/mod.rs index 10a5d50fd..cfeded80d 100644 --- a/ecstore/src/config/mod.rs +++ b/ecstore/src/config/mod.rs @@ -19,6 +19,14 @@ lazy_static! { pub static ref GLOBAL_ConfigSys: ConfigSys = ConfigSys::new(); } +/// Standard config keys and values. +pub const ENABLE_KEY: &str = "enable"; +pub const COMMENT_KEY: &str = "comment"; + +/// Enable values +pub const ENABLE_ON: &str = "on"; +pub const ENABLE_OFF: &str = "off"; + pub const ENV_ACCESS_KEY: &str = "RUSTFS_ACCESS_KEY"; pub const ENV_SECRET_KEY: &str = "RUSTFS_SECRET_KEY"; pub const ENV_ROOT_USER: &str = "RUSTFS_ROOT_USER"; diff --git a/rustfs/src/event.rs b/rustfs/src/event.rs index 1ef9db356..dbbe11e42 100644 --- a/rustfs/src/event.rs +++ b/rustfs/src/event.rs @@ -11,6 +11,7 @@ pub(crate) async fn init_event_notifier(notifier_config: Option) { } else { info!("event_config is empty"); // rustfs_event::get_event_notifier_config().clone() + NotifierConfig::default() }; info!("using event_config: {:?}", config); diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 4e598a92c..aa34f9616 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -510,7 +510,7 @@ async fn run(opt: config::Opt) -> Result<()> { GLOBAL_ConfigSys.init(store.clone()).await?; // event system configuration - // GLOBAL_EventSys.init(store.clone()).await?; + // GLOBAL_EVENT_SYS.init(store.clone()).await?; // Initialize event notifier event::init_event_notifier(opt.event_config).await;