diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index de9dac41f..bd4709247 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -271,14 +271,18 @@ impl DefaultAdminUsecase { } pub fn execute_collect_dependency_readiness(&self) -> DependencyReadiness { - if let Some(context) = &self.context { - let _ = context.object_store(); - let _ = context.iam(); - } + let iam_ready = self + .context + .as_ref() + .map(|context| { + let _ = context.object_store(); + context.iam().is_ready() + }) + .unwrap_or(false); DependencyReadiness { storage_ready: new_object_layer_fn().is_some(), - iam_ready: rustfs_iam::get().is_ok(), + iam_ready, } } } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 62043bb38..085555b02 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -15,7 +15,7 @@ //! Bucket application use-case contracts. #![allow(dead_code)] -use crate::app::context::{AppContext, get_global_app_context}; +use crate::app::context::{AppContext, default_notify_interface, get_global_app_context}; use crate::auth::get_condition_values; use crate::error::ApiError; use crate::server::RemoteAddr; @@ -46,7 +46,6 @@ use rustfs_ecstore::client::object_api_utils::to_s3s_etag; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI}; -use rustfs_notify::notifier_global; use rustfs_policy::policy::{ action::{Action, S3Action}, {BucketPolicy, BucketPolicyArgs, Effect, Validator}, @@ -1189,7 +1188,12 @@ impl DefaultBucketUsecase { .map_err(ApiError::from)?; let region = resolve_notification_region(rustfs_ecstore::global::get_global_region(), request_region); - let clear_rules = notifier_global::clear_bucket_notification_rules(&bucket); + let notify = self + .context + .as_ref() + .map(|context| context.notify()) + .unwrap_or_else(default_notify_interface); + let clear_rules = notify.clear_bucket_notification_rules(&bucket); let parse_rules = async { let mut event_rules = Vec::new(); @@ -1223,7 +1227,8 @@ impl DefaultBucketUsecase { event_rules_result.map_err(|e| s3_error!(InvalidArgument, "Invalid ARN in notification configuration: {e}"))?; warn!("notify event rules: {:?}", &event_rules); - notifier_global::add_event_specific_rules(&bucket, ®ion, &event_rules) + notify + .add_event_specific_rules(&bucket, ®ion, &event_rules) .await .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?; diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 10ae85e86..0d97b8e26 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -17,14 +17,18 @@ //! for storage, IAM, and KMS handles. #![allow(dead_code)] +use async_trait::async_trait; use rustfs_ecstore::store::ECStore; use rustfs_iam::{store::object::ObjectStore, sys::IamSys}; use rustfs_kms::KmsServiceManager; +use rustfs_notify::{EventArgs, NotificationError, notifier_global}; +use rustfs_targets::{EventName, arn::TargetID}; use std::sync::{Arc, OnceLock}; /// IAM interface for application-layer use-cases. pub trait IamInterface: Send + Sync { fn handle(&self) -> Arc>; + fn is_ready(&self) -> bool; } /// KMS interface for application-layer use-cases. @@ -32,6 +36,21 @@ pub trait KmsInterface: Send + Sync { fn handle(&self) -> Arc; } +/// Notify interface for application-layer use-cases. +#[async_trait] +pub trait NotifyInterface: Send + Sync { + async fn notify(&self, args: EventArgs); + + async fn add_event_specific_rules( + &self, + bucket_name: &str, + region: &str, + event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), NotificationError>; + + async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError>; +} + /// Default IAM interface adapter. pub struct IamHandle { iam: Arc>, @@ -47,6 +66,10 @@ impl IamInterface for IamHandle { fn handle(&self) -> Arc> { self.iam.clone() } + + fn is_ready(&self) -> bool { + rustfs_iam::get().is_ok() + } } /// Default KMS interface adapter. @@ -66,17 +89,47 @@ impl KmsInterface for KmsHandle { } } +/// Default notify interface adapter. +#[derive(Default)] +pub struct NotifyHandle; + +#[async_trait] +impl NotifyInterface for NotifyHandle { + async fn notify(&self, args: EventArgs) { + notifier_global::notify(args).await; + } + + async fn add_event_specific_rules( + &self, + bucket_name: &str, + region: &str, + event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), NotificationError> { + notifier_global::add_event_specific_rules(bucket_name, region, event_rules).await + } + + async fn clear_bucket_notification_rules(&self, bucket_name: &str) -> Result<(), NotificationError> { + notifier_global::clear_bucket_notification_rules(bucket_name).await + } +} + /// Application-layer context with explicit dependencies. #[derive(Clone)] pub struct AppContext { object_store: Arc, iam: Arc, kms: Arc, + notify: Arc, } impl AppContext { pub fn new(object_store: Arc, iam: Arc, kms: Arc) -> Self { - Self { object_store, iam, kms } + Self { + object_store, + iam, + kms, + notify: default_notify_interface(), + } } pub fn with_default_interfaces( @@ -98,6 +151,14 @@ impl AppContext { pub fn kms(&self) -> Arc { self.kms.clone() } + + pub fn notify(&self) -> Arc { + self.notify.clone() + } +} + +pub fn default_notify_interface() -> Arc { + Arc::new(NotifyHandle) } static GLOBAL_APP_CONTEXT: OnceLock> = OnceLock::new(); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index b1051abb9..b81235821 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -15,7 +15,7 @@ //! Object application use-case contracts. #![allow(dead_code)] -use crate::app::context::{AppContext, get_global_app_context}; +use crate::app::context::{AppContext, default_notify_interface, get_global_app_context}; use crate::config::workload_profiles::get_global_buffer_config; use crate::error::ApiError; use crate::storage::access::{ReqInfo, authorize_request, has_bypass_governance_header}; @@ -72,7 +72,7 @@ use rustfs_filemeta::{ REPLICATE_INCOMING_DELETE, ReplicationStatusType, ReplicationType, RestoreStatusOps, VersionPurgeStatusType, parse_restore_obj_status, }; -use rustfs_notify::{EventArgsBuilder, notifier_global}; +use rustfs_notify::EventArgsBuilder; use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_rio::{CompressReader, DecryptReader, EncryptReader, EtagReader, HardLimitReader, HashReader, Reader, WarpReader}; use rustfs_s3select_api::{ @@ -2647,6 +2647,11 @@ impl DefaultObjectUsecase { } let req_headers = req.headers.clone(); + let notify = self + .context + .as_ref() + .map(|context| context.notify()) + .unwrap_or_else(default_notify_interface); tokio::spawn(async move { for res in delete_results { if let Some(dobj) = res.delete_object { @@ -2671,7 +2676,7 @@ impl DefaultObjectUsecase { .user_agent(get_request_user_agent(&req_headers)) .build(); - notifier_global::notify(event_args).await; + notify.notify(event_args).await; } } });