Merge branch 'main' of github.com:rustfs/s3-rustfs into feature/observability-metrics

# Conflicts:
#	crates/notify/src/event.rs
#	crates/notify/src/global.rs
#	crates/notify/src/integration.rs
#	crates/notify/src/notifier.rs

crates/notify/src/arn.rs

@@ -41,7 +41,7 @@ impl TargetID {
         ARN {
             target_id: self.clone(),
             region: region.to_string(),
-            service: DEFAULT_ARN_SERVICE.to_string(), // Default Service
+            service: DEFAULT_ARN_SERVICE.to_string(),     // Default Service
             partition: DEFAULT_ARN_PARTITION.to_string(), // Default partition
         }
     }
@@ -112,7 +112,7 @@ impl ARN {
         ARN {
             target_id,
             region,
-            service: DEFAULT_ARN_SERVICE.to_string(), // Default is sqs
+            service: DEFAULT_ARN_SERVICE.to_string(),     // Default is sqs
             partition: DEFAULT_ARN_PARTITION.to_string(), // Default is rustfs partition
         }
     }
@@ -121,16 +121,10 @@ impl ARN {
     /// Returns the ARN string in the format "{ARN_PREFIX}:{region}:{target_id}"
     #[allow(clippy::inherent_to_string)]
     pub fn to_arn_string(&self) -> String {
-        if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty()
-        {
+        if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty() {
             return String::new();
         }
-        format!(
-            "{}:{}:{}",
-            ARN_PREFIX,
-            self.region,
-            self.target_id.to_id_string()
-        )
+        format!("{}:{}:{}", ARN_PREFIX, self.region, self.target_id.to_id_string())
     }
 
     /// Parsing ARN from string
@@ -162,8 +156,7 @@ impl ARN {
 
 impl fmt::Display for ARN {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty()
-        {
+        if self.target_id.id.is_empty() && self.target_id.name.is_empty() && self.region.is_empty() {
             // Returns an empty string if all parts are empty
             return Ok(());
         }

crates/notify/src/error.rs

@@ -1,6 +1,6 @@
+use crate::arn::TargetID;
 use std::io;
 use thiserror::Error;
-use crate::arn::TargetID;
 
 /// Error types for the store
 #[derive(Debug, Error)]

crates/notify/src/factory.rs

@@ -1,7 +1,7 @@
 use crate::store::DEFAULT_LIMIT;
 use crate::{
     error::TargetError,
-    target::{mqtt::MQTTArgs, webhook::WebhookArgs, Target},
+    target::{Target, mqtt::MQTTArgs, webhook::WebhookArgs},
 };
 use async_trait::async_trait;
 use ecstore::config::{ENABLE_KEY, ENABLE_ON, KVS};

crates/notify/src/integration.rs

@@ -1,15 +1,15 @@
 use crate::arn::TargetID;
 use crate::store::{Key, Store};
 use crate::{
-    error::NotificationError, notifier::EventNotifier, registry::TargetRegistry, rules::BucketNotificationConfig, stream, Event, EventName,
-    StoreError, Target,
+    Event, EventName, StoreError, Target, error::NotificationError, notifier::EventNotifier, registry::TargetRegistry,
+    rules::BucketNotificationConfig, stream,
 };
 use ecstore::config::{Config, KVS};
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::{Duration, Instant};
-use tokio::sync::{mpsc, RwLock, Semaphore};
+use tokio::sync::{RwLock, Semaphore, mpsc};
 use tracing::{debug, error, info, warn};
 
 /// Notify the system of monitoring indicators

crates/notify/src/lib.rs

@@ -26,7 +26,7 @@ pub use rules::BucketNotificationConfig;
 use std::io::IsTerminal;
 pub use target::Target;
 
-use tracing_subscriber::{fmt, prelude::*, util::SubscriberInitExt, EnvFilter};
+use tracing_subscriber::{EnvFilter, fmt, prelude::*, util::SubscriberInitExt};
 
 /// Initialize the tracing log system
 ///

crates/notify/src/notifier.rs

@@ -1,5 +1,5 @@
 use crate::arn::TargetID;
-use crate::{error::NotificationError, event::Event, rules::RulesMap, target::Target, EventName};
+use crate::{EventName, error::NotificationError, event::Event, rules::RulesMap, target::Target};
 use dashmap::DashMap;
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::RwLock;

crates/notify/src/rules/config.rs

@@ -1,11 +1,11 @@
 use super::rules_map::RulesMap;
 // Keep for existing structure if any, or remove if not used
 use super::xml_config::ParseConfigError as BucketNotificationConfigError;
+use crate::EventName;
 use crate::arn::TargetID;
+use crate::rules::NotificationConfiguration;
 use crate::rules::pattern_rules;
 use crate::rules::target_id_set;
-use crate::rules::NotificationConfiguration;
-use crate::EventName;
 use std::collections::HashMap;
 use std::io::Read;
 

crates/notify/src/rules/pattern.rs

@@ -78,10 +78,7 @@ mod tests {
         assert_eq!(new_pattern(Some(""), Some("b")), "*b");
         assert_eq!(new_pattern(None, None), "");
         assert_eq!(new_pattern(Some("prefix"), Some("suffix")), "prefix*suffix");
-        assert_eq!(
-            new_pattern(Some("prefix/"), Some("/suffix")),
-            "prefix/*suffix"
-        ); // prefix/* + */suffix -> prefix/**/suffix -> prefix/*/suffix
+        assert_eq!(new_pattern(Some("prefix/"), Some("/suffix")), "prefix/*suffix"); // prefix/* + */suffix -> prefix/**/suffix -> prefix/*/suffix
     }
 
     #[test]

crates/notify/src/rules/pattern_rules.rs

@@ -23,9 +23,7 @@ impl PatternRules {
 
     /// Checks if there are any rules that match the given object name.
     pub fn match_simple(&self, object_name: &str) -> bool {
-        self.rules
-            .keys()
-            .any(|p| pattern::match_simple(p, object_name))
+        self.rules.keys().any(|p| pattern::match_simple(p, object_name))
     }
 
     /// Returns all TargetIDs that match the object name.
@@ -61,8 +59,7 @@ impl PatternRules {
         for (pattern, self_targets) in &self.rules {
             match other.rules.get(pattern) {
                 Some(other_targets) => {
-                    let diff_targets: TargetIdSet =
-                        self_targets.difference(other_targets).cloned().collect();
+                    let diff_targets: TargetIdSet = self_targets.difference(other_targets).cloned().collect();
                     if !diff_targets.is_empty() {
                         result_rules.insert(pattern.clone(), diff_targets);
                     }
@@ -73,8 +70,6 @@ impl PatternRules {
                 }
             }
         }
-        PatternRules {
-            rules: result_rules,
-        }
+        PatternRules { rules: result_rules }
     }
 }

crates/notify/src/rules/xml_config.rs

@@ -1,5 +1,5 @@
 use super::pattern;
-use crate::arn::{ArnError, TargetIDError, ARN};
+use crate::arn::{ARN, ArnError, TargetIDError};
 use crate::event::EventName;
 use serde::{Deserialize, Serialize};
 use std::collections::HashSet;

crates/notify/src/store.rs

@@ -1,5 +1,5 @@
 use crate::error::StoreError;
-use serde::{de::DeserializeOwned, Serialize};
+use serde::{Serialize, de::DeserializeOwned};
 use snap::raw::{Decoder, Encoder};
 use std::sync::{Arc, RwLock};
 use std::{

crates/notify/src/stream.rs

@@ -1,13 +1,13 @@
 use crate::{
-    error::TargetError, integration::NotificationMetrics,
+    Event, StoreError,
+    error::TargetError,
+    integration::NotificationMetrics,
     store::{Key, Store},
     target::Target,
-    Event,
-    StoreError,
 };
 use std::sync::Arc;
 use std::time::{Duration, Instant};
-use tokio::sync::{mpsc, Semaphore};
+use tokio::sync::{Semaphore, mpsc};
 use tokio::time::sleep;
 use tracing::{debug, error, info, warn};
 

crates/notify/src/target/mod.rs

@@ -1,8 +1,8 @@
-use std::sync::Arc;
 use crate::arn::TargetID;
 use crate::store::{Key, Store};
 use crate::{Event, StoreError, TargetError};
 use async_trait::async_trait;
+use std::sync::Arc;
 
 pub mod mqtt;
 pub mod webhook;

crates/notify/src/target/mqtt.rs

@@ -1,22 +1,22 @@
 use crate::store::{Key, STORE_EXTENSION};
 use crate::target::ChannelTargetType;
 use crate::{
-    arn::TargetID, error::TargetError,
+    StoreError, Target,
+    arn::TargetID,
+    error::TargetError,
     event::{Event, EventLog},
     store::Store,
-    StoreError,
-    Target,
 };
 use async_trait::async_trait;
-use rumqttc::{mqttbytes::Error as MqttBytesError, ConnectionError};
 use rumqttc::{AsyncClient, EventLoop, MqttOptions, Outgoing, Packet, QoS};
+use rumqttc::{ConnectionError, mqttbytes::Error as MqttBytesError};
 use std::sync::Arc;
 use std::{
     path::PathBuf,
     sync::atomic::{AtomicBool, Ordering},
     time::Duration,
 };
-use tokio::sync::{mpsc, Mutex, OnceCell};
+use tokio::sync::{Mutex, OnceCell, mpsc};
 use tracing::{debug, error, info, instrument, trace, warn};
 use url::Url;
 use urlencoding;

crates/notify/src/target/webhook.rs

@@ -1,19 +1,19 @@
 use crate::store::STORE_EXTENSION;
 use crate::target::ChannelTargetType;
 use crate::{
-    arn::TargetID, error::TargetError,
+    StoreError, Target,
+    arn::TargetID,
+    error::TargetError,
     event::{Event, EventLog},
     store::{Key, Store},
-    StoreError,
-    Target,
 };
 use async_trait::async_trait;
 use reqwest::{Client, StatusCode, Url};
 use std::{
     path::PathBuf,
     sync::{
-        atomic::{AtomicBool, Ordering},
         Arc,
+        atomic::{AtomicBool, Ordering},
     },
     time::Duration,
 };