diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 74435dee3..69183f0d9 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -20,13 +20,13 @@ pub mod target; // Re-exports pub use error::{NotificationError, StoreError, TargetError}; pub use event::{Event, EventArgs, EventLog, EventName}; -pub use global::{initialize, notification_system}; +pub use global::{initialize, is_notification_system_initialized, notification_system}; pub use integration::NotificationSystem; pub use rules::BucketNotificationConfig; use std::io::IsTerminal; pub use target::Target; -use tracing_subscriber::{EnvFilter, fmt, prelude::*, util::SubscriberInitExt}; +use tracing_subscriber::{fmt, prelude::*, util::SubscriberInitExt, EnvFilter}; /// Initialize the tracing log system /// diff --git a/rustfs/src/event.rs b/rustfs/src/event.rs index 6d5712cdf..24617fb49 100644 --- a/rustfs/src/event.rs +++ b/rustfs/src/event.rs @@ -40,3 +40,25 @@ pub(crate) async fn init_event_notifier() { } }); } + +/// Shuts down the event notifier system gracefully +pub async fn shutdown_event_notifier() { + info!("Shutting down event notifier system..."); + + if !rustfs_notify::is_notification_system_initialized() { + info!("Event notifier system is not initialized, nothing to shut down."); + return; + } + + let system = match rustfs_notify::notification_system() { + Some(sys) => sys, + None => { + error!("Event notifier system is not initialized."); + return; + } + }; + + // Call the shutdown function from the rustfs_notify module + system.shutdown().await; + info!("Event notifier system shut down successfully."); +} diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index c1fa1ce20..68f17cdfc 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -14,6 +14,7 @@ mod storage; use crate::auth::IAMAuth; use crate::console::{init_console_cfg, CONSOLE_CONFIG}; // Ensure the correct path for parse_license is imported +use crate::event::shutdown_event_notifier; use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT}; use bytes::Bytes; use chrono::Datelike; @@ -324,14 +325,13 @@ async fn run(opt: config::Opt) -> Result<()> { }; #[cfg(not(unix))] - let (mut sigterm_future, mut sigint_future) = { + let (mut sigterm_inner, mut sigint_inner) = { // Windows platform uses Ctrl+C as the only signal source // Create two never-finished futures as placeholders + info!("Running on Windows, only Ctrl+C signal will be handled"); (std::pin::pin!(std::future::pending::<()>()), std::pin::pin!(std::future::pending::<()>())) }; - let mut sigterm_inner = sigterm_inner; - let mut sigint_inner = sigint_inner; let hybrid_service = TowerToHyperService::new( tower::ServiceBuilder::new() .layer( @@ -402,24 +402,25 @@ async fn run(opt: config::Opt) -> Result<()> { } } } + _ = ctrl_c.as_mut() => { info!("Ctrl-C received in worker thread"); let _ = shutdown_tx_clone.send(()); break; } - #[cfg(unix)] - _ = sigint_inner.recv() => { - info!("SIGINT received in worker thread"); - let _ = shutdown_tx_clone.send(()); - break; - } - #[cfg(unix)] - _ = sigterm_inner.recv() => { - info!("SIGTERM received in worker thread"); - let _ = shutdown_tx_clone.send(()); - break; - } + Some(_) = sigint_inner.recv() => { + info!("SIGINT received in worker thread"); + let _ = shutdown_tx_clone.send(()); + break; + } + + Some(_) = sigterm_inner.recv() => { + info!("SIGTERM received in worker thread"); + let _ = shutdown_tx_clone.send(()); + break; + } + _ = shutdown_rx.recv() => { info!("Shutdown signal received in worker thread"); break; @@ -592,14 +593,8 @@ async fn run(opt: config::Opt) -> Result<()> { // update the status to stopping first state_manager.update(ServiceState::Stopping); - // // Stop the notification system - // if rustfs_event::is_ready() { - // // stop event notifier - // rustfs_event::shutdown().await.map_err(|err| { - // error!("Failed to shut down the notification system: {}", err); - // Error::from_string(err.to_string()) - // })?; - // } + // Stop the notification system + shutdown_event_notifier().await; info!("Server is stopping..."); let _ = shutdown_tx.send(());