From 0f22b21c8d32f666b4f1f8f2daa74d26accdadd7 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 21 May 2025 16:44:59 +0800 Subject: [PATCH] improve channel adapter --- Cargo.toml | 1 + crates/event/Cargo.toml | 2 +- crates/event/examples/full.rs | 5 +- crates/event/examples/simple.rs | 6 +- crates/event/src/adapter/kafka.rs | 4 +- crates/event/src/adapter/mod.rs | 47 +++ crates/event/src/adapter/mqtt.rs | 4 +- crates/event/src/adapter/webhook.rs | 13 +- crates/event/src/bus.rs | 10 - crates/event/src/config.rs | 11 +- crates/event/src/event_sys/mod.rs | 267 ------------------ crates/event/src/lib.rs | 13 +- crates/event/src/{store.rs => store/event.rs} | 0 crates/event/src/store/mod.rs | 119 ++++++++ crates/event/tests/integration.rs | 6 +- ecstore/src/config/com.rs | 89 +++--- ecstore/src/config/storageclass.rs | 3 +- rustfs/src/event.rs | 46 +-- rustfs/src/main.rs | 2 + .../storage/{event_notifier.rs => event.rs} | 0 rustfs/src/storage/mod.rs | 2 +- 21 files changed, 278 insertions(+), 372 deletions(-) delete mode 100644 crates/event/src/event_sys/mod.rs rename crates/event/src/{store.rs => store/event.rs} (100%) create mode 100644 crates/event/src/store/mod.rs rename rustfs/src/storage/{event_notifier.rs => event.rs} (100%) diff --git a/Cargo.toml b/Cargo.toml index ab7ec0921..4b0e1af84 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -74,6 +74,7 @@ datafusion = "46.0.1" derive_builder = "0.20.2" dioxus = { version = "0.6.3", features = ["router"] } dirs = "6.0.0" +dotenvy = "0.15.7" flatbuffers = "25.2.10" futures = "0.3.31" futures-core = "0.3.31" diff --git a/crates/event/Cargo.toml b/crates/event/Cargo.toml index 8aa232761..752ea7faa 100644 --- a/crates/event/Cargo.toml +++ b/crates/event/Cargo.toml @@ -40,7 +40,7 @@ rdkafka = { workspace = true, features = ["tokio"], optional = true } tokio = { workspace = true, features = ["test-util"] } tracing-subscriber = { workspace = true } axum = { workspace = true } -dotenvy = "0.15.7" +dotenvy = { workspace = true } [lints] workspace = true diff --git a/crates/event/examples/full.rs b/crates/event/examples/full.rs index 128dfe2d8..c9e17af94 100644 --- a/crates/event/examples/full.rs +++ b/crates/event/examples/full.rs @@ -1,5 +1,6 @@ use rustfs_event::{ - AdapterConfig, Bucket, Error as NotifierError, Event, Identity, Metadata, Name, NotifierConfig, Object, Source, WebhookConfig, + AdapterConfig, Bucket, ChannelAdapterType, Error as NotifierError, Event, Identity, Metadata, Name, NotifierConfig, Object, + Source, WebhookConfig, }; use std::collections::HashMap; use tokio::signal; @@ -103,7 +104,7 @@ async fn main() -> Result<(), Box> { }) .s3(metadata) .source(source) - .channels(vec!["webhook".to_string()]) + .channels(vec![ChannelAdapterType::Webhook.to_string()]) .build() .expect("failed to create event"); diff --git a/crates/event/examples/simple.rs b/crates/event/examples/simple.rs index 5160186ee..84cc32c12 100644 --- a/crates/event/examples/simple.rs +++ b/crates/event/examples/simple.rs @@ -1,5 +1,5 @@ -use rustfs_event::create_adapters; use rustfs_event::NotifierSystem; +use rustfs_event::{create_adapters, ChannelAdapterType}; use rustfs_event::{AdapterConfig, NotifierConfig, WebhookConfig}; use rustfs_event::{Bucket, Event, Identity, Metadata, Name, Object, Source}; use std::collections::HashMap; @@ -31,7 +31,7 @@ async fn main() -> Result<(), Box> { // event_load_config // loading configuration from environment variables - let _config = NotifierConfig::event_load_config(Some("./crates/event-notifier/examples/event.toml".to_string())); + let _config = NotifierConfig::event_load_config(Some("./crates/event/examples/event.toml".to_string())); tracing::info!("event_load_config config: {:?} \n", _config); dotenvy::dotenv()?; let _config = NotifierConfig::event_load_config(None); @@ -77,7 +77,7 @@ async fn main() -> Result<(), Box> { }) .s3(metadata) .source(source) - .channels(vec!["webhook".to_string()]) + .channels(vec![ChannelAdapterType::Webhook.to_string()]) .build() .expect("failed to create event"); diff --git a/crates/event/src/adapter/kafka.rs b/crates/event/src/adapter/kafka.rs index 0abd1f00c..1ad0ddbd3 100644 --- a/crates/event/src/adapter/kafka.rs +++ b/crates/event/src/adapter/kafka.rs @@ -1,7 +1,7 @@ -use crate::ChannelAdapter; use crate::Error; use crate::Event; use crate::KafkaConfig; +use crate::{ChannelAdapter, ChannelAdapterType}; use async_trait::async_trait; use rdkafka::error::KafkaError; use rdkafka::producer::{FutureProducer, FutureRecord}; @@ -60,7 +60,7 @@ impl KafkaAdapter { #[async_trait] impl ChannelAdapter for KafkaAdapter { fn name(&self) -> String { - "kafka".to_string() + ChannelAdapterType::Kafka.to_string() } async fn send(&self, event: &Event) -> Result<(), Error> { diff --git a/crates/event/src/adapter/mod.rs b/crates/event/src/adapter/mod.rs index 426fd2d83..a2d6471c2 100644 --- a/crates/event/src/adapter/mod.rs +++ b/crates/event/src/adapter/mod.rs @@ -11,6 +11,53 @@ pub(crate) mod mqtt; #[cfg(feature = "webhook")] pub(crate) mod 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. +/// +/// # Variants +/// +/// - `Webhook`: Represents a webhook adapter. +/// - `Kafka`: Represents a Kafka adapter. +/// - `Mqtt`: Represents an MQTT adapter. +/// +/// # Example +/// +/// ``` +/// use rustfs_event::ChannelAdapterType; +/// +/// let adapter_type = ChannelAdapterType::Webhook; +/// match adapter_type { +/// ChannelAdapterType::Webhook => println!("Using webhook adapter"), +/// ChannelAdapterType::Kafka => println!("Using Kafka adapter"), +/// ChannelAdapterType::Mqtt => println!("Using MQTT adapter"), +/// } +pub enum ChannelAdapterType { + Webhook, + Kafka, + Mqtt, +} + +impl ChannelAdapterType { + pub fn as_str(&self) -> &'static str { + match self { + ChannelAdapterType::Webhook => "webhook", + ChannelAdapterType::Kafka => "kafka", + ChannelAdapterType::Mqtt => "mqtt", + } + } +} + +impl std::fmt::Display for ChannelAdapterType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ChannelAdapterType::Webhook => write!(f, "webhook"), + ChannelAdapterType::Kafka => write!(f, "kafka"), + ChannelAdapterType::Mqtt => write!(f, "mqtt"), + } + } +} + /// The `ChannelAdapter` trait defines the interface for all channel adapters. #[async_trait] pub trait ChannelAdapter: Send + Sync + 'static { diff --git a/crates/event/src/adapter/mqtt.rs b/crates/event/src/adapter/mqtt.rs index 9aab61e8e..ffd3b4583 100644 --- a/crates/event/src/adapter/mqtt.rs +++ b/crates/event/src/adapter/mqtt.rs @@ -1,7 +1,7 @@ -use crate::ChannelAdapter; use crate::Error; use crate::Event; use crate::MqttConfig; +use crate::{ChannelAdapter, ChannelAdapterType}; use async_trait::async_trait; use rumqttc::{AsyncClient, MqttOptions, QoS}; use std::time::Duration; @@ -33,7 +33,7 @@ impl MqttAdapter { #[async_trait] impl ChannelAdapter for MqttAdapter { fn name(&self) -> String { - "mqtt".to_string() + ChannelAdapterType::Mqtt.to_string() } async fn send(&self, event: &Event) -> Result<(), Error> { diff --git a/crates/event/src/adapter/webhook.rs b/crates/event/src/adapter/webhook.rs index 447c463ee..4cf523bbb 100644 --- a/crates/event/src/adapter/webhook.rs +++ b/crates/event/src/adapter/webhook.rs @@ -1,7 +1,7 @@ -use crate::ChannelAdapter; use crate::Error; use crate::Event; use crate::WebhookConfig; +use crate::{ChannelAdapter, ChannelAdapterType}; use async_trait::async_trait; use reqwest::{Client, RequestBuilder}; use std::time::Duration; @@ -16,10 +16,11 @@ pub struct WebhookAdapter { impl WebhookAdapter { /// Creates a new Webhook adapter. pub fn new(config: WebhookConfig) -> Self { - let client = Client::builder() - .timeout(Duration::from_secs(config.timeout)) - .build() - .expect("Failed to build reqwest client"); + let mut builder = Client::builder(); + if config.timeout > 0 { + builder = builder.timeout(Duration::from_secs(config.timeout)); + } + let client = builder.build().expect("Failed to build reqwest client"); Self { config, client } } /// Builds the request to send the event. @@ -40,7 +41,7 @@ impl WebhookAdapter { #[async_trait] impl ChannelAdapter for WebhookAdapter { fn name(&self) -> String { - "webhook".to_string() + ChannelAdapterType::Webhook.to_string() } async fn send(&self, event: &Event) -> Result<(), Error> { diff --git a/crates/event/src/bus.rs b/crates/event/src/bus.rs index 5cabfc22e..c68a6a1d0 100644 --- a/crates/event/src/bus.rs +++ b/crates/event/src/bus.rs @@ -21,17 +21,10 @@ pub async fn event_bus( shutdown: CancellationToken, shutdown_complete: Option>, ) -> Result<(), Error> { - let mut current_log = Log { - event_name: crate::event::Name::Everything, - key: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs().to_string(), - records: Vec::new(), - }; - let mut unprocessed_events = Vec::new(); loop { tokio::select! { Some(event) = rx.recv() => { - current_log.records.push(event.clone()); let mut send_tasks = Vec::new(); for adapter in &adapters { if event.channels.contains(&adapter.name()) { @@ -54,9 +47,6 @@ pub async fn event_bus( unprocessed_events.push(failed_event); } } - - // Clear the current log because we only care about unprocessed events - current_log.records.clear(); } _ = shutdown.cancelled() => { tracing::info!("Shutting down event bus, saving pending logs..."); diff --git a/crates/event/src/config.rs b/crates/event/src/config.rs index 12b2c7a03..996db2d3d 100644 --- a/crates/event/src/config.rs +++ b/crates/event/src/config.rs @@ -3,6 +3,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::env; +const DEFAULT_CONFIG_FILE: &str = "event"; + /// Configuration for the webhook adapter. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct WebhookConfig { @@ -206,17 +208,18 @@ impl NotifierConfig { } } -const DEFAULT_CONFIG_FILE: &str = "event"; - /// Provide temporary directories as default storage paths fn default_store_path() -> String { env::var("EVENT_STORE_PATH").unwrap_or_else(|e| { - tracing::error!("Failed to get EVENT_STORE_PATH: {}", e); + tracing::info!("Failed to get `EVENT_STORE_PATH` 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 fn default_channel_capacity() -> usize { - 10000 // Reasonable default values for high concurrency systems + env::var("EVENT_CHANNEL_CAPACITY") + .unwrap_or_else(|_| "10000".to_string()) + .parse() + .unwrap_or(10000) // Default to 10000 if parsing fails } diff --git a/crates/event/src/event_sys/mod.rs b/crates/event/src/event_sys/mod.rs deleted file mode 100644 index 41f76e39e..000000000 --- a/crates/event/src/event_sys/mod.rs +++ /dev/null @@ -1,267 +0,0 @@ -use crate::NotifierConfig; -use common::error::{Error, Result}; -use ecstore::config::com::CONFIG_PREFIX; -use ecstore::config::error::{is_err_config_not_found, ConfigError}; -use ecstore::disk::RUSTFS_META_BUCKET; -use ecstore::store::ECStore; -use ecstore::store_api::{ObjectInfo, ObjectOptions, PutObjReader}; -use ecstore::store_err::is_err_object_not_found; -use ecstore::utils::path::SLASH_SEPARATOR; -use ecstore::StorageAPI; -use http::HeaderMap; -use lazy_static::lazy_static; -use std::io::Cursor; -use std::sync::{Arc, OnceLock}; -use tracing::{error, instrument, warn}; - -lazy_static! { - pub static ref GLOBAL_EventSys: EventSys = EventSys::new(); - pub static ref GLOBAL_EventSysConfig: OnceLock = OnceLock::new(); -} -/// * config file -const CONFIG_FILE: &str = "event.json"; - -/// event sys config -const EVENT: &str = "event"; - -#[derive(Debug)] -pub struct EventSys {} - -impl Default for EventSys { - fn default() -> Self { - Self::new() - } -} - -impl EventSys { - pub fn new() -> Self { - Self {} - } - #[instrument(skip_all)] - pub async fn init(&self, api: Arc) -> Result<()> { - tracing::info!("event sys config init start"); - let cfg = read_config_without_migrate(api.clone().clone()).await?; - let _ = GLOBAL_EventSysConfig.set(cfg); - tracing::info!("event sys config init done"); - Ok(()) - } -} - -/// get event sys config file -/// -/// # Returns -/// NotifierConfig -pub fn get_event_notifier_config() -> &'static NotifierConfig { - GLOBAL_EventSysConfig.get_or_init(NotifierConfig::default) -} - -fn get_event_sys_file() -> String { - format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE) -} - -/// read config without migrate -/// -/// # Parameters -/// - `api`: StorageAPI -/// -/// # Returns -/// Configuration information -pub async fn read_config_without_migrate(api: Arc) -> Result { - let config_file = get_event_sys_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) - } - } - }; - - read_server_config(api, data.as_slice()).await -} - -/// save config with options -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `file`: file name -/// - `data`: data to save -/// - `opts`: object options -/// -/// # Returns -/// Result -pub async fn save_config_with_opts(api: Arc, file: &str, data: Vec, opts: &ObjectOptions) -> Result<()> { - let size = data.len(); - let _ = api - .put_object(RUSTFS_META_BUCKET, file, &mut PutObjReader::new(Box::new(Cursor::new(data)), size), opts) - .await?; - Ok(()) -} - -/// new server config -/// -/// # Returns -/// NotifierConfig -fn new_server_config() -> NotifierConfig { - NotifierConfig::new() -} - -async fn new_and_save_server_config(api: Arc) -> Result { - let cfg = new_server_config(); - save_server_config(api, &cfg).await?; - - Ok(cfg) -} - -async fn read_server_config(api: Arc, data: &[u8]) -> Result { - let cfg = { - if data.is_empty() { - let config_file = get_event_sys_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) - } - } - }; - // TODO: decrypt - - NotifierConfig::unmarshal(cfg_data.as_slice())? - } else { - NotifierConfig::unmarshal(data)? - } - }; - - Ok(cfg.merge()) -} - -/// save server config -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `cfg`: configuration to save -/// -/// # Returns -/// Result -async fn save_server_config(api: Arc, cfg: &NotifierConfig) -> Result<()> { - let data = cfg.marshal()?; - - let config_file = get_event_sys_file(); - - save_config(api, &config_file, data).await -} - -/// save config -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `file`: file name -/// - `data`: data to save -/// -/// # Returns -/// Result -pub async fn save_config(api: Arc, file: &str, data: Vec) -> Result<()> { - save_config_with_opts( - api, - file, - data, - &ObjectOptions { - max_parity: true, - ..Default::default() - }, - ) - .await -} - -/// delete config -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `file`: file name -/// -/// # Returns -/// Result -pub async fn delete_config(api: Arc, file: &str) -> Result<()> { - match api - .delete_object( - RUSTFS_META_BUCKET, - file, - ObjectOptions { - delete_prefix: true, - delete_prefix_object: true, - ..Default::default() - }, - ) - .await - { - Ok(_) => Ok(()), - Err(err) => { - if is_err_object_not_found(&err) { - Err(Error::new(ConfigError::NotFound)) - } else { - Err(err) - } - } - } -} - -/// read config -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `file`: file name -/// -/// # Returns -/// Configuration data -pub async fn read_config(api: Arc, file: &str) -> Result> { - let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?; - Ok(data) -} - -/// read config with metadata -/// -/// # Parameters -/// - `api`: StorageAPI -/// - `file`: file name -/// - `opts`: object options -/// -/// # Returns -/// Configuration data and object info -pub async fn read_config_with_metadata( - api: Arc, - file: &str, - opts: &ObjectOptions, -) -> Result<(Vec, ObjectInfo)> { - let h = HeaderMap::new(); - let mut rd = api - .get_object_reader(RUSTFS_META_BUCKET, file, None, h, opts) - .await - .map_err(|err| { - if is_err_object_not_found(&err) { - Error::new(ConfigError::NotFound) - } else { - err - } - })?; - - let data = rd.read_all().await?; - - if data.is_empty() { - return Err(Error::new(ConfigError::NotFound)); - } - - Ok((data, rd.object_info)) -} diff --git a/crates/event/src/lib.rs b/crates/event/src/lib.rs index 3b5de3fb1..20b576dc6 100644 --- a/crates/event/src/lib.rs +++ b/crates/event/src/lib.rs @@ -3,7 +3,6 @@ mod bus; mod config; mod error; mod event; -mod event_sys; mod global; mod notifier; mod store; @@ -16,6 +15,7 @@ pub use adapter::mqtt::MqttAdapter; #[cfg(feature = "webhook")] pub use adapter::webhook::WebhookAdapter; pub use adapter::ChannelAdapter; +pub use adapter::ChannelAdapterType; pub use bus::event_bus; #[cfg(all(feature = "kafka", target_os = "linux"))] pub use config::KafkaConfig; @@ -29,12 +29,9 @@ 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::EventStore; +pub use store::event::EventStore; -pub use event_sys::delete_config; -pub use event_sys::get_event_notifier_config; -pub use event_sys::read_config; -pub use event_sys::save_config; +pub use store::get_event_notifier_config; -pub use event_sys::EventSys; -pub use event_sys::GLOBAL_EventSys; +pub use store::EventSys; +pub use store::GLOBAL_EventSys; diff --git a/crates/event/src/store.rs b/crates/event/src/store/event.rs similarity index 100% rename from crates/event/src/store.rs rename to crates/event/src/store/event.rs diff --git a/crates/event/src/store/mod.rs b/crates/event/src/store/mod.rs new file mode 100644 index 000000000..d1cec06d9 --- /dev/null +++ b/crates/event/src/store/mod.rs @@ -0,0 +1,119 @@ +pub(crate) mod event; + +use crate::NotifierConfig; +use common::error::Result; +use ecstore::config::com::{read_config, save_config, CONFIG_PREFIX}; +use ecstore::config::error::is_err_config_not_found; +use ecstore::store::ECStore; +use ecstore::utils::path::SLASH_SEPARATOR; +use ecstore::StorageAPI; +use lazy_static::lazy_static; +use std::sync::{Arc, OnceLock}; +use tracing::{error, instrument, warn}; + +lazy_static! { + pub static ref GLOBAL_EventSys: EventSys = EventSys::new(); + pub static ref GLOBAL_EventSysConfig: OnceLock = OnceLock::new(); +} +/// * config file +const CONFIG_FILE: &str = "event.json"; + +/// event sys config +const EVENT: &str = "event"; + +#[derive(Debug)] +pub struct EventSys {} + +impl Default for EventSys { + fn default() -> Self { + Self::new() + } +} + +impl EventSys { + pub fn new() -> Self { + Self {} + } + #[instrument(skip_all)] + pub async fn init(&self, api: Arc) -> Result<()> { + tracing::info!("event sys config init start"); + let cfg = read_config_without_migrate(api.clone().clone()).await?; + let _ = GLOBAL_EventSysConfig.set(cfg); + tracing::info!("event sys config init done"); + Ok(()) + } +} + +/// get event sys config file +/// +/// # Returns +/// NotifierConfig +pub fn get_event_notifier_config() -> &'static NotifierConfig { + GLOBAL_EventSysConfig.get_or_init(NotifierConfig::default) +} + +fn get_event_sys_file() -> String { + format!("{}{}{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, EVENT, SLASH_SEPARATOR, CONFIG_FILE) +} + +/// new server config +/// +/// # Returns +/// NotifierConfig +fn new_server_config() -> NotifierConfig { + NotifierConfig::new() +} + +async fn new_and_save_server_config(api: Arc) -> Result { + let cfg = new_server_config(); + save_server_config(api, &cfg).await?; + + Ok(cfg) +} + +/// save server config +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `cfg`: configuration to save +/// +/// # Returns +/// Result +async fn save_server_config(api: Arc, cfg: &NotifierConfig) -> Result<()> { + let data = cfg.marshal()?; + + let config_file = get_event_sys_file(); + + save_config(api, &config_file, data).await +} + +/// read config without migrate +/// +/// # Parameters +/// - `api`: StorageAPI +/// +/// # Returns +/// Configuration information +pub async fn read_config_without_migrate(api: Arc) -> Result { + let config_file = get_event_sys_file(); + let data = match read_config(api.clone(), config_file.as_str()).await { + Ok(data) => { + if data.is_empty() { + return new_and_save_server_config(api).await; + } + data + } + Err(err) if is_err_config_not_found(&err) => { + warn!("config not found, start to init"); + return new_and_save_server_config(api).await; + } + Err(err) => { + error!("read config error: {:?}", err); + return Err(err); + } + }; + + // TODO: decrypt if needed + let cfg = NotifierConfig::unmarshal(data.as_slice())?; + Ok(cfg.merge()) +} diff --git a/crates/event/tests/integration.rs b/crates/event/tests/integration.rs index 9d8972382..73166fb30 100644 --- a/crates/event/tests/integration.rs +++ b/crates/event/tests/integration.rs @@ -1,4 +1,4 @@ -use rustfs_event::{AdapterConfig, NotifierSystem, WebhookConfig}; +use rustfs_event::{AdapterConfig, ChannelAdapterType, NotifierSystem, WebhookConfig}; use rustfs_event::{Bucket, Event, EventBuilder, Identity, Metadata, Name, Object, Source}; use rustfs_event::{ChannelAdapter, WebhookAdapter}; use std::collections::HashMap; @@ -57,7 +57,7 @@ async fn test_webhook_adapter() { .response_elements(HashMap::new()) .s3(metadata) .source(source) - .channels(vec!["webhook".to_string()]) + .channels(vec![ChannelAdapterType::Webhook.to_string()]) .build() .expect("failed to create event"); @@ -122,7 +122,7 @@ async fn test_notification_system() { principal_id: "user123".to_string(), }) .event_time("2023-10-01T12:00:00.000Z") - .channels(vec!["webhook".to_string()]) + .channels(vec![ChannelAdapterType::Webhook.to_string()]) .build() .expect("failed to create event"); diff --git a/ecstore/src/config/com.rs b/ecstore/src/config/com.rs index ab34ebee5..17482997e 100644 --- a/ecstore/src/config/com.rs +++ b/ecstore/src/config/com.rs @@ -33,33 +33,41 @@ lazy_static! { }; } -/// * read config +/// Reads configuration file content from the storage system /// -/// * @param api -/// * @param file +/// This function reads a specified configuration file through the provided storage API interface +/// and returns its raw byte content. It's a basic configuration reading function that returns +/// only the configuration data without any metadata. +/// +/// # Parameters +/// * `api` - An Arc smart pointer containing an implementation of the StorageAPI trait +/// * `file` - The name of the configuration file to read +/// +/// # Returns +/// * `Result>` - Returns the configuration file contents as bytes on success, or an error on failure +/// +/// # Errors +/// May return the following errors: +/// * `ConfigError::NotFound` - When the requested configuration file does not exist +/// * Other storage operation related errors /// -/// * @return -/// * @description -/// * read config -/// * @error -/// * * ConfigError::NotFound pub async fn read_config(api: Arc, file: &str) -> Result> { let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?; Ok(data) } -/// * read_config_with_metadata -/// read config with metadata +/// read config with metadata with api,file and opts /// -/// * @param api -/// * @param file -/// * @param opts +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// - `opts`: object options /// -/// * @return -/// * @description -/// * read config with metadata -/// * @error -/// * * ConfigError::NotFound +/// # Returns +/// Result<(Vec, ObjectInfo)> +/// +/// # Errors +/// - ConfigError::NotFound pub async fn read_config_with_metadata( api: Arc, file: &str, @@ -86,15 +94,16 @@ pub async fn read_config_with_metadata( Ok((data, rd.object_info)) } -/// * save_config +/// save config with api,file and data /// -/// * @param api -/// * @param file -/// * @param data +/// # Parameters /// -/// * @return -/// * @description -/// * save config +/// - `api`: StorageAPI +/// - `file`: file name +/// - `data`: data to save +/// +/// # Returns +/// Result pub async fn save_config(api: Arc, file: &str, data: Vec) -> Result<()> { save_config_with_opts( api, @@ -108,13 +117,15 @@ pub async fn save_config(api: Arc, file: &str, data: Vec) .await } -/// * delete_config +/// delete config with api and file +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// +/// # Returns +/// Result /// -/// * @param api -/// * @param file -/// * @return -/// * @description -/// * delete config pub async fn delete_config(api: Arc, file: &str) -> Result<()> { match api .delete_object( @@ -139,15 +150,15 @@ pub async fn delete_config(api: Arc, file: &str) -> Result<()> } } -/// * save_config_with_opts /// save config with opts -/// * @param api -/// * @param file -/// * @param data -/// * @param opts -/// * @return -/// * @description -/// * save config with opts +/// +/// # Parameters +/// - `api`: StorageAPI +/// - `file`: file name +/// - `data`: data to save +/// +/// # Returns +/// Result pub async fn save_config_with_opts(api: Arc, file: &str, data: Vec, opts: &ObjectOptions) -> Result<()> { let size = data.len(); let _ = api diff --git a/ecstore/src/config/storageclass.rs b/ecstore/src/config/storageclass.rs index ef5199fc6..943054b2a 100644 --- a/ecstore/src/config/storageclass.rs +++ b/ecstore/src/config/storageclass.rs @@ -8,7 +8,8 @@ use lazy_static::lazy_static; use serde::{Deserialize, Serialize}; use tracing::warn; -// default_parity_count 默认配置，根据磁盘总数分配校验磁盘数量 +/// Default parity count for a given drive count +/// The default configuration allocates the number of check disks based on the total number of disks pub fn default_parity_count(drive: usize) -> usize { match drive { 1 => 0, diff --git a/rustfs/src/event.rs b/rustfs/src/event.rs index e64f12822..00650f868 100644 --- a/rustfs/src/event.rs +++ b/rustfs/src/event.rs @@ -3,29 +3,29 @@ use tracing::{error, info, instrument}; #[instrument] pub(crate) async fn init_event_notifier(notifier_config: Option) { - // Initialize event notifier - if notifier_config.is_some() { - info!("event_config is not empty"); - tokio::spawn(async move { - let config = NotifierConfig::event_load_config(notifier_config); - let result = rustfs_event::initialize(&config).await; - if let Err(e) = result { - error!("Failed to initialize event notifier: {}", e); - } else { - info!("Event notifier initialized successfully"); - } - }); + info!("Initializing event notifier..."); + let notifier_config_present = notifier_config.is_some(); + let config = if notifier_config_present { + info!("event_config is not empty, path: {:?}", notifier_config); + NotifierConfig::event_load_config(notifier_config) } else { info!("event_config is empty"); - tokio::spawn(async move { - let config = rustfs_event::get_event_notifier_config(); - info!("event_config is {:?}", config); - let result = rustfs_event::initialize(config).await; - if let Err(e) = result { - error!("Failed to initialize event notifier: {}", e); - } else { - info!("Event notifier initialized successfully"); - } - }); - } + rustfs_event::get_event_notifier_config().clone() + }; + + info!("using event_config: {:?}", config); + tokio::spawn(async move { + let result = rustfs_event::initialize(&config).await; + match result { + Ok(_) => info!( + "event notifier initialized successfully {}", + if notifier_config_present { + "by config file" + } else { + "by sys config" + } + ), + Err(e) => error!("Failed to initialize event notifier: {}", e), + } + }); } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index e3b867385..9a6fbd7aa 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -506,8 +506,10 @@ async fn run(opt: config::Opt) -> Result<()> { })?; ecconfig::init(); + // config system configuration GLOBAL_ConfigSys.init(store.clone()).await?; + // event system configuration GLOBAL_EventSys.init(store.clone()).await?; // Initialize event notifier diff --git a/rustfs/src/storage/event_notifier.rs b/rustfs/src/storage/event.rs similarity index 100% rename from rustfs/src/storage/event_notifier.rs rename to rustfs/src/storage/event.rs diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 2f8ec8b88..df31b3f05 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -1,5 +1,5 @@ pub mod access; pub mod ecfs; pub mod error; -mod event_notifier; +mod event; pub mod options;