mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-10 23:26:53 +00:00
feat(targets): extract targets module into a standalone crate (#441)
* init audit logger module * add audit webhook default config kvs * feat: Add comprehensive tests for authentication module (#309) * feat: add comprehensive tests for authentication module - Add 33 unit tests covering all public functions in auth.rs - Test IAMAuth struct creation and secret key validation - Test check_claims_from_token with various credential types and scenarios - Test session token extraction from headers and query parameters - Test condition values generation for different user types - Test query parameter parsing with edge cases - Test Credentials helper methods (is_expired, is_temp, is_service_account) - Ensure tests handle global state dependencies gracefully - All tests pass successfully with 100% coverage of testable functions * style: fix code formatting issues * Add verification script for checking PR branch statuses and tests Co-authored-by: anzhengchao <anzhengchao@gmail.com> * fix: resolve clippy uninlined format args warning --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> * feat: add basic tests for core storage module (#313) * feat: add basic tests for core storage module - Add 6 unit tests for FS struct and basic functionality - Test FS creation, Debug and Clone trait implementations - Test RUSTFS_OWNER constant definition and values - Test S3 error code creation and handling - Test compression format detection for common file types - Include comprehensive documentation about integration test needs Note: Full S3 API testing requires complex setup with storage backend, global configuration, and network infrastructure - better suited for integration tests rather than unit tests. * style: fix code formatting issues * fix: resolve clippy warnings in storage tests --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> * feat: add tests for admin handlers module (#314) * feat: add tests for admin handlers module - Add 5 new unit tests for admin handler functionality - Test AccountInfo struct creation, serialization and default values - Test creation of all admin handler structs (13 handlers) - Test HealOpts JSON serialization and deserialization - Test HealOpts URL encoding/decoding with proper field types - Maintain existing test while adding comprehensive coverage - Include documentation about integration test requirements All tests pass successfully with proper error handling for complex dependencies. * style: fix code formatting issues * fix: resolve clippy warnings in admin handlers tests --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> * build(deps): bump the dependencies group with 3 updates (#326) * perf: avoid transmitting parity shards when the object is good (#322) * upgrade version * Fix: fix data integrity check Signed-off-by: junxiang Mu <1948535941@qq.com> * Fix: Separate Clippy's fix and check commands into two commands. Signed-off-by: junxiang Mu <1948535941@qq.com> * fix: miss inline metadata (#345) * Update dependabot.yml * fix: Fixed an issue where the list_objects_v2 API did not return dire… (#352) * fix: Fixed an issue where the list_objects_v2 API did not return directory names when they conflicted with file names in the same bucket (e.g., test/ vs. test.txt, aaa/ vs. aaa.csv) (#335) * fix: adjusted the order of directory listings * init * fix * fix * feat: add docker usage for rustfs mcp (#365) * feat: enhance metadata extraction with object name for MIME type detection Signed-off-by: junxiang Mu <1948535941@qq.com> * Feature: lock support auto release Signed-off-by: junxiang Mu <1948535941@qq.com> * improve lock Signed-off-by: junxiang Mu <1948535941@qq.com> * Fix: fix scanner detect Signed-off-by: junxiang Mu <1948535941@qq.com> * Fix: clippy && fmt Signed-off-by: junxiang Mu <1948535941@qq.com> * refactor(ecstore): Optimize memory usage for object integrity verification Change the object integrity verification from reading all data to streaming processing to avoid memory overflow caused by large objects. Modify the TLS key log check to use environment variables directly instead of configuration constants. Add memory limits for object data reading in the AHM module. Signed-off-by: junxiang Mu <1948535941@qq.com> * Chore: reduce PR template checklist Signed-off-by: junxiang Mu <1948535941@qq.com> * Chore: remove comment code (#376) Signed-off-by: junxiang Mu <1948535941@qq.com> * chore: upgrade actions/checkout from v4 to v5 (#381) * chore: upgrade actions/checkout from v4 to v5 - Update GitHub Actions checkout action version - Ensure compatibility with latest workflow features - Maintain existing checkout behavior and configuration * upgrade version * fix * add and improve code for notify * feat: extend rustfs mcp with bucket creation and deletion (#416) * feat: extend rustfs mcp with bucket creation and deletion * update file to fix pipeline error * change variable name to fix pipeline error * fix(ecstore): add async-recursion to resolve nightly trait solver reg… (#415) * fix(ecstore): add async-recursion to resolve nightly trait solver regression The newest nightly compiler switched to the new trait solver, which currently rejects async recursive functions that were previously accepted. This causes the following compilation failures: - `LocalDisk::delete_file()` - `LocalDisk::scan_dir()` Add `async-recursion` as a workspace dependency and annotate both functions with `#[async_recursion]` so that the crate compiles cleanly with the latest nightly and will continue to build once the new solver lands in stable. Signed-off-by: reigadegr <2722688642@qq.com> * fix: resolve duplicate bound error in scan_dir function Replaced inline trait bounds with where clause to avoid duplication caused by macro expansion. Signed-off-by: reigadegr <2722688642@qq.com> --------- Signed-off-by: reigadegr <2722688642@qq.com> Co-authored-by: 安正超 <anzhengchao@gmail.com> * fix:make bucket exists (#428) * feat: include user-defined metadata in S3 response (#431) * fix: simplify Docker entrypoint following efficient user switching pattern (#421) * fix: simplify Docker entrypoint following efficient user switching pattern - Remove ALL file permission modifications (no chown at all) - Use chroot --userspec or gosu to switch user context - Extremely simple and fast implementation - Zero filesystem modifications for permissions Fixes #388 * Update entrypoint.sh Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update entrypoint.sh Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update entrypoint.sh Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * wip * wip * wip --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * docs: update doc/docker-data-dir README.md (#432) * add targets crates * feat(targets): extract targets module into a standalone crate - Move all target-related code (MQTT, Webhook, etc.) into a new `targets` crate - Update imports and dependencies to reference the new crate - Refactor interfaces to ensure compatibility with the new crate structure - Adjust Cargo.toml and workspace configuration accordingly * fix * fix --------- Signed-off-by: junxiang Mu <1948535941@qq.com> Signed-off-by: reigadegr <2722688642@qq.com> Co-authored-by: 安正超 <anzhengchao@gmail.com> Co-authored-by: Cursor Agent <cursoragent@cursor.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: zzhpro <56196563+zzhpro@users.noreply.github.com> Co-authored-by: junxiang Mu <1948535941@qq.com> Co-authored-by: weisd <im@weisd.in> Co-authored-by: shiro.lee <69624924+shiroleeee@users.noreply.github.com> Co-authored-by: majinghe <42570491+majinghe@users.noreply.github.com> Co-authored-by: guojidan <63799833+guojidan@users.noreply.github.com> Co-authored-by: reigadegr <103645642+reigadegr@users.noreply.github.com> Co-authored-by: 0xdx2 <xuedamon2@gmail.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
houseme
@@ -1,242 +0,0 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::TargetError;
|
||||
use rustfs_config::notify::{ARN_PREFIX, DEFAULT_ARN_PARTITION, DEFAULT_ARN_SERVICE};
|
||||
use serde::{Deserialize, Deserializer, Serialize, Serializer};
|
||||
use std::fmt;
|
||||
use std::str::FromStr;
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum TargetIDError {
|
||||
#[error("Invalid TargetID format '{0}', expect 'ID:Name'")]
|
||||
InvalidFormat(String),
|
||||
}
|
||||
|
||||
/// Target ID, used to identify notification targets
|
||||
#[derive(Debug, Clone, Eq, PartialEq, Hash, PartialOrd, Ord)]
|
||||
pub struct TargetID {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
}
|
||||
|
||||
impl TargetID {
|
||||
pub fn new(id: String, name: String) -> Self {
|
||||
Self { id, name }
|
||||
}
|
||||
|
||||
/// Convert to string representation
|
||||
pub fn to_id_string(&self) -> String {
|
||||
format!("{}:{}", self.id, self.name)
|
||||
}
|
||||
|
||||
/// Create an ARN
|
||||
pub fn to_arn(&self, region: &str) -> ARN {
|
||||
ARN {
|
||||
target_id: self.clone(),
|
||||
region: region.to_string(),
|
||||
service: DEFAULT_ARN_SERVICE.to_string(), // Default Service
|
||||
partition: DEFAULT_ARN_PARTITION.to_string(), // Default partition
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for TargetID {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}:{}", self.id, self.name)
|
||||
}
|
||||
}
|
||||
|
||||
impl FromStr for TargetID {
|
||||
type Err = TargetIDError;
|
||||
|
||||
fn from_str(s: &str) -> Result<Self, Self::Err> {
|
||||
let parts: Vec<&str> = s.splitn(2, ':').collect();
|
||||
if parts.len() == 2 {
|
||||
Ok(TargetID {
|
||||
id: parts[0].to_string(),
|
||||
name: parts[1].to_string(),
|
||||
})
|
||||
} else {
|
||||
Err(TargetIDError::InvalidFormat(s.to_string()))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Serialize for TargetID {
|
||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
serializer.serialize_str(&self.to_id_string())
|
||||
}
|
||||
}
|
||||
|
||||
impl<'de> Deserialize<'de> for TargetID {
|
||||
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let s = String::deserialize(deserializer)?;
|
||||
TargetID::from_str(&s).map_err(serde::de::Error::custom)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ArnError {
|
||||
#[error("Invalid ARN format '{0}'")]
|
||||
InvalidFormat(String),
|
||||
#[error("ARN component missing")]
|
||||
MissingComponents,
|
||||
}
|
||||
|
||||
/// ARN - AWS resource name representation
|
||||
#[derive(Debug, Clone, Eq, PartialEq)]
|
||||
pub struct ARN {
|
||||
pub target_id: TargetID,
|
||||
pub region: String,
|
||||
// Service types, such as "sqs", "sns", "lambda", etc. This defaults to "sqs" to match the Go example.
|
||||
pub service: String,
|
||||
// Partitions such as "aws", "aws-cn", or customizations such as "rustfs", etc.
|
||||
pub partition: String,
|
||||
}
|
||||
|
||||
impl ARN {
|
||||
pub fn new(target_id: TargetID, region: String) -> Self {
|
||||
ARN {
|
||||
target_id,
|
||||
region,
|
||||
service: DEFAULT_ARN_SERVICE.to_string(), // Default is sqs
|
||||
partition: DEFAULT_ARN_PARTITION.to_string(), // Default is rustfs partition
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the string representation of 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() {
|
||||
return String::new();
|
||||
}
|
||||
format!("{}:{}:{}", ARN_PREFIX, self.region, self.target_id.to_id_string())
|
||||
}
|
||||
|
||||
/// Parsing ARN from string
|
||||
pub fn parse(s: &str) -> Result<Self, TargetError> {
|
||||
if !s.starts_with(ARN_PREFIX) {
|
||||
return Err(TargetError::InvalidARN(s.to_string()));
|
||||
}
|
||||
|
||||
let tokens: Vec<&str> = s.split(':').collect();
|
||||
if tokens.len() != 6 {
|
||||
return Err(TargetError::InvalidARN(s.to_string()));
|
||||
}
|
||||
|
||||
if tokens[4].is_empty() || tokens[5].is_empty() {
|
||||
return Err(TargetError::InvalidARN(s.to_string()));
|
||||
}
|
||||
|
||||
Ok(ARN {
|
||||
region: tokens[3].to_string(),
|
||||
target_id: TargetID {
|
||||
id: tokens[4].to_string(),
|
||||
name: tokens[5].to_string(),
|
||||
},
|
||||
service: tokens[2].to_string(), // Service Type
|
||||
partition: tokens[1].to_string(), // Partition
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
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() {
|
||||
// Returns an empty string if all parts are empty
|
||||
return Ok(());
|
||||
}
|
||||
write!(
|
||||
f,
|
||||
"arn:{}:{}:{}:{}:{}",
|
||||
self.partition, self.service, self.region, self.target_id.id, self.target_id.name
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
impl FromStr for ARN {
|
||||
type Err = ArnError;
|
||||
|
||||
fn from_str(s: &str) -> Result<Self, Self::Err> {
|
||||
let parts: Vec<&str> = s.split(':').collect();
|
||||
if parts.len() < 6 {
|
||||
return Err(ArnError::InvalidFormat(s.to_string()));
|
||||
}
|
||||
|
||||
if parts[0] != "arn" {
|
||||
return Err(ArnError::InvalidFormat(s.to_string()));
|
||||
}
|
||||
|
||||
let partition = parts[1].to_string();
|
||||
let service = parts[2].to_string();
|
||||
let region = parts[3].to_string();
|
||||
let id = parts[4].to_string();
|
||||
let name = parts[5..].join(":"); // The name section may contain colons, although this is not usually the case in SQS ARN
|
||||
|
||||
if id.is_empty() || name.is_empty() {
|
||||
return Err(ArnError::MissingComponents);
|
||||
}
|
||||
|
||||
Ok(ARN {
|
||||
target_id: TargetID { id, name },
|
||||
region,
|
||||
service,
|
||||
partition,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Serialization implementation
|
||||
impl Serialize for ARN {
|
||||
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
|
||||
where
|
||||
S: Serializer,
|
||||
{
|
||||
serializer.serialize_str(&self.to_arn_string())
|
||||
}
|
||||
}
|
||||
|
||||
impl<'de> Deserialize<'de> for ARN {
|
||||
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
// deserializer.deserialize_str(ARNVisitor)
|
||||
let s = String::deserialize(deserializer)?;
|
||||
if s.is_empty() {
|
||||
// Handle an empty ARN string, for example, creating an empty or default Arn instance
|
||||
// Or return an error based on business logic
|
||||
// Here we create an empty TargetID and region Arn
|
||||
return Ok(ARN {
|
||||
target_id: TargetID {
|
||||
id: String::new(),
|
||||
name: String::new(),
|
||||
},
|
||||
region: String::new(),
|
||||
service: DEFAULT_ARN_SERVICE.to_string(),
|
||||
partition: DEFAULT_ARN_PARTITION.to_string(),
|
||||
});
|
||||
}
|
||||
ARN::from_str(&s).map_err(serde::de::Error::custom)
|
||||
}
|
||||
}
|
||||
+12
-94
@@ -1,98 +1,22 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::arn::TargetID;
|
||||
use rustfs_targets::TargetError;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use std::io;
|
||||
use thiserror::Error;
|
||||
|
||||
/// Error types for the store
|
||||
#[derive(Debug, Error)]
|
||||
pub enum StoreError {
|
||||
#[error("I/O error: {0}")]
|
||||
Io(#[from] io::Error),
|
||||
|
||||
#[error("Serialization error: {0}")]
|
||||
Serialization(String),
|
||||
|
||||
#[error("Deserialization error: {0}")]
|
||||
Deserialization(String),
|
||||
|
||||
#[error("Compression error: {0}")]
|
||||
Compression(String),
|
||||
|
||||
#[error("Entry limit exceeded")]
|
||||
LimitExceeded,
|
||||
|
||||
#[error("Entry not found")]
|
||||
NotFound,
|
||||
|
||||
#[error("Invalid entry: {0}")]
|
||||
Internal(String), // Added internal error type
|
||||
}
|
||||
|
||||
/// Error types for targets
|
||||
#[derive(Debug, Error)]
|
||||
pub enum TargetError {
|
||||
#[error("Storage error: {0}")]
|
||||
Storage(String),
|
||||
|
||||
#[error("Network error: {0}")]
|
||||
Network(String),
|
||||
|
||||
#[error("Request error: {0}")]
|
||||
Request(String),
|
||||
|
||||
#[error("Timeout error: {0}")]
|
||||
Timeout(String),
|
||||
|
||||
#[error("Authentication error: {0}")]
|
||||
Authentication(String),
|
||||
|
||||
#[error("Configuration error: {0}")]
|
||||
Configuration(String),
|
||||
|
||||
#[error("Encoding error: {0}")]
|
||||
Encoding(String),
|
||||
|
||||
#[error("Serialization error: {0}")]
|
||||
Serialization(String),
|
||||
|
||||
#[error("Target not connected")]
|
||||
NotConnected,
|
||||
|
||||
#[error("Target initialization failed: {0}")]
|
||||
Initialization(String),
|
||||
|
||||
#[error("Invalid ARN: {0}")]
|
||||
InvalidARN(String),
|
||||
|
||||
#[error("Unknown error: {0}")]
|
||||
Unknown(String),
|
||||
|
||||
#[error("Target is disabled")]
|
||||
Disabled,
|
||||
|
||||
#[error("Configuration parsing error: {0}")]
|
||||
ParseError(String),
|
||||
|
||||
#[error("Failed to save configuration: {0}")]
|
||||
SaveConfig(String),
|
||||
|
||||
#[error("Server not initialized: {0}")]
|
||||
ServerNotInitialized(String),
|
||||
}
|
||||
|
||||
/// Error types for the notification system
|
||||
#[derive(Debug, Error)]
|
||||
pub enum NotificationError {
|
||||
@@ -135,9 +59,3 @@ pub enum NotificationError {
|
||||
#[error("Server not initialized")]
|
||||
ServerNotInitialized,
|
||||
}
|
||||
|
||||
impl From<url::ParseError> for TargetError {
|
||||
fn from(err: url::ParseError) -> Self {
|
||||
TargetError::Configuration(format!("URL parse error: {err}"))
|
||||
}
|
||||
}
|
||||
|
||||
+1
-286
@@ -13,285 +13,11 @@
|
||||
// limitations under the License.
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use rustfs_targets::EventName;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
use std::fmt;
|
||||
use url::form_urlencoded;
|
||||
|
||||
/// Error returned when parsing event name string fails。
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct ParseEventNameError(String);
|
||||
|
||||
impl fmt::Display for ParseEventNameError {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "Invalid event name:{}", self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for ParseEventNameError {}
|
||||
|
||||
/// Represents the type of event that occurs on the object.
|
||||
/// Based on AWS S3 event type and includes RustFS extension.
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
|
||||
pub enum EventName {
|
||||
// Single event type (values are 1-32 for compatible mask logic)
|
||||
ObjectAccessedGet = 1,
|
||||
ObjectAccessedGetRetention = 2,
|
||||
ObjectAccessedGetLegalHold = 3,
|
||||
ObjectAccessedHead = 4,
|
||||
ObjectAccessedAttributes = 5,
|
||||
ObjectCreatedCompleteMultipartUpload = 6,
|
||||
ObjectCreatedCopy = 7,
|
||||
ObjectCreatedPost = 8,
|
||||
ObjectCreatedPut = 9,
|
||||
ObjectCreatedPutRetention = 10,
|
||||
ObjectCreatedPutLegalHold = 11,
|
||||
ObjectCreatedPutTagging = 12,
|
||||
ObjectCreatedDeleteTagging = 13,
|
||||
ObjectRemovedDelete = 14,
|
||||
ObjectRemovedDeleteMarkerCreated = 15,
|
||||
ObjectRemovedDeleteAllVersions = 16,
|
||||
ObjectRemovedNoOP = 17,
|
||||
BucketCreated = 18,
|
||||
BucketRemoved = 19,
|
||||
ObjectReplicationFailed = 20,
|
||||
ObjectReplicationComplete = 21,
|
||||
ObjectReplicationMissedThreshold = 22,
|
||||
ObjectReplicationReplicatedAfterThreshold = 23,
|
||||
ObjectReplicationNotTracked = 24,
|
||||
ObjectRestorePost = 25,
|
||||
ObjectRestoreCompleted = 26,
|
||||
ObjectTransitionFailed = 27,
|
||||
ObjectTransitionComplete = 28,
|
||||
ScannerManyVersions = 29, // ObjectManyVersions corresponding to Go
|
||||
ScannerLargeVersions = 30, // ObjectLargeVersions corresponding to Go
|
||||
ScannerBigPrefix = 31, // PrefixManyFolders corresponding to Go
|
||||
LifecycleDelMarkerExpirationDelete = 32, // ILMDelMarkerExpirationDelete corresponding to Go
|
||||
|
||||
// Compound "All" event type (no sequential value for mask)
|
||||
ObjectAccessedAll,
|
||||
ObjectCreatedAll,
|
||||
ObjectRemovedAll,
|
||||
ObjectReplicationAll,
|
||||
ObjectRestoreAll,
|
||||
ObjectTransitionAll,
|
||||
ObjectScannerAll, // New, from Go
|
||||
Everything, // New, from Go
|
||||
}
|
||||
|
||||
// Single event type sequential array for Everything.expand()
|
||||
const SINGLE_EVENT_NAMES_IN_ORDER: [EventName; 32] = [
|
||||
EventName::ObjectAccessedGet,
|
||||
EventName::ObjectAccessedGetRetention,
|
||||
EventName::ObjectAccessedGetLegalHold,
|
||||
EventName::ObjectAccessedHead,
|
||||
EventName::ObjectAccessedAttributes,
|
||||
EventName::ObjectCreatedCompleteMultipartUpload,
|
||||
EventName::ObjectCreatedCopy,
|
||||
EventName::ObjectCreatedPost,
|
||||
EventName::ObjectCreatedPut,
|
||||
EventName::ObjectCreatedPutRetention,
|
||||
EventName::ObjectCreatedPutLegalHold,
|
||||
EventName::ObjectCreatedPutTagging,
|
||||
EventName::ObjectCreatedDeleteTagging,
|
||||
EventName::ObjectRemovedDelete,
|
||||
EventName::ObjectRemovedDeleteMarkerCreated,
|
||||
EventName::ObjectRemovedDeleteAllVersions,
|
||||
EventName::ObjectRemovedNoOP,
|
||||
EventName::BucketCreated,
|
||||
EventName::BucketRemoved,
|
||||
EventName::ObjectReplicationFailed,
|
||||
EventName::ObjectReplicationComplete,
|
||||
EventName::ObjectReplicationMissedThreshold,
|
||||
EventName::ObjectReplicationReplicatedAfterThreshold,
|
||||
EventName::ObjectReplicationNotTracked,
|
||||
EventName::ObjectRestorePost,
|
||||
EventName::ObjectRestoreCompleted,
|
||||
EventName::ObjectTransitionFailed,
|
||||
EventName::ObjectTransitionComplete,
|
||||
EventName::ScannerManyVersions,
|
||||
EventName::ScannerLargeVersions,
|
||||
EventName::ScannerBigPrefix,
|
||||
EventName::LifecycleDelMarkerExpirationDelete,
|
||||
];
|
||||
|
||||
const LAST_SINGLE_TYPE_VALUE: u32 = EventName::LifecycleDelMarkerExpirationDelete as u32;
|
||||
|
||||
impl EventName {
|
||||
/// The parsed string is EventName.
|
||||
pub fn parse(s: &str) -> Result<Self, ParseEventNameError> {
|
||||
match s {
|
||||
"s3:BucketCreated:*" => Ok(EventName::BucketCreated),
|
||||
"s3:BucketRemoved:*" => Ok(EventName::BucketRemoved),
|
||||
"s3:ObjectAccessed:*" => Ok(EventName::ObjectAccessedAll),
|
||||
"s3:ObjectAccessed:Get" => Ok(EventName::ObjectAccessedGet),
|
||||
"s3:ObjectAccessed:GetRetention" => Ok(EventName::ObjectAccessedGetRetention),
|
||||
"s3:ObjectAccessed:GetLegalHold" => Ok(EventName::ObjectAccessedGetLegalHold),
|
||||
"s3:ObjectAccessed:Head" => Ok(EventName::ObjectAccessedHead),
|
||||
"s3:ObjectAccessed:Attributes" => Ok(EventName::ObjectAccessedAttributes),
|
||||
"s3:ObjectCreated:*" => Ok(EventName::ObjectCreatedAll),
|
||||
"s3:ObjectCreated:CompleteMultipartUpload" => Ok(EventName::ObjectCreatedCompleteMultipartUpload),
|
||||
"s3:ObjectCreated:Copy" => Ok(EventName::ObjectCreatedCopy),
|
||||
"s3:ObjectCreated:Post" => Ok(EventName::ObjectCreatedPost),
|
||||
"s3:ObjectCreated:Put" => Ok(EventName::ObjectCreatedPut),
|
||||
"s3:ObjectCreated:PutRetention" => Ok(EventName::ObjectCreatedPutRetention),
|
||||
"s3:ObjectCreated:PutLegalHold" => Ok(EventName::ObjectCreatedPutLegalHold),
|
||||
"s3:ObjectCreated:PutTagging" => Ok(EventName::ObjectCreatedPutTagging),
|
||||
"s3:ObjectCreated:DeleteTagging" => Ok(EventName::ObjectCreatedDeleteTagging),
|
||||
"s3:ObjectRemoved:*" => Ok(EventName::ObjectRemovedAll),
|
||||
"s3:ObjectRemoved:Delete" => Ok(EventName::ObjectRemovedDelete),
|
||||
"s3:ObjectRemoved:DeleteMarkerCreated" => Ok(EventName::ObjectRemovedDeleteMarkerCreated),
|
||||
"s3:ObjectRemoved:NoOP" => Ok(EventName::ObjectRemovedNoOP),
|
||||
"s3:ObjectRemoved:DeleteAllVersions" => Ok(EventName::ObjectRemovedDeleteAllVersions),
|
||||
"s3:LifecycleDelMarkerExpiration:Delete" => Ok(EventName::LifecycleDelMarkerExpirationDelete),
|
||||
"s3:Replication:*" => Ok(EventName::ObjectReplicationAll),
|
||||
"s3:Replication:OperationFailedReplication" => Ok(EventName::ObjectReplicationFailed),
|
||||
"s3:Replication:OperationCompletedReplication" => Ok(EventName::ObjectReplicationComplete),
|
||||
"s3:Replication:OperationMissedThreshold" => Ok(EventName::ObjectReplicationMissedThreshold),
|
||||
"s3:Replication:OperationReplicatedAfterThreshold" => Ok(EventName::ObjectReplicationReplicatedAfterThreshold),
|
||||
"s3:Replication:OperationNotTracked" => Ok(EventName::ObjectReplicationNotTracked),
|
||||
"s3:ObjectRestore:*" => Ok(EventName::ObjectRestoreAll),
|
||||
"s3:ObjectRestore:Post" => Ok(EventName::ObjectRestorePost),
|
||||
"s3:ObjectRestore:Completed" => Ok(EventName::ObjectRestoreCompleted),
|
||||
"s3:ObjectTransition:Failed" => Ok(EventName::ObjectTransitionFailed),
|
||||
"s3:ObjectTransition:Complete" => Ok(EventName::ObjectTransitionComplete),
|
||||
"s3:ObjectTransition:*" => Ok(EventName::ObjectTransitionAll),
|
||||
"s3:Scanner:ManyVersions" => Ok(EventName::ScannerManyVersions),
|
||||
"s3:Scanner:LargeVersions" => Ok(EventName::ScannerLargeVersions),
|
||||
"s3:Scanner:BigPrefix" => Ok(EventName::ScannerBigPrefix),
|
||||
// ObjectScannerAll and Everything cannot be parsed from strings, because the Go version also does not define their string representation.
|
||||
_ => Err(ParseEventNameError(s.to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a string representation of the event type.
|
||||
pub fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
EventName::BucketCreated => "s3:BucketCreated:*",
|
||||
EventName::BucketRemoved => "s3:BucketRemoved:*",
|
||||
EventName::ObjectAccessedAll => "s3:ObjectAccessed:*",
|
||||
EventName::ObjectAccessedGet => "s3:ObjectAccessed:Get",
|
||||
EventName::ObjectAccessedGetRetention => "s3:ObjectAccessed:GetRetention",
|
||||
EventName::ObjectAccessedGetLegalHold => "s3:ObjectAccessed:GetLegalHold",
|
||||
EventName::ObjectAccessedHead => "s3:ObjectAccessed:Head",
|
||||
EventName::ObjectAccessedAttributes => "s3:ObjectAccessed:Attributes",
|
||||
EventName::ObjectCreatedAll => "s3:ObjectCreated:*",
|
||||
EventName::ObjectCreatedCompleteMultipartUpload => "s3:ObjectCreated:CompleteMultipartUpload",
|
||||
EventName::ObjectCreatedCopy => "s3:ObjectCreated:Copy",
|
||||
EventName::ObjectCreatedPost => "s3:ObjectCreated:Post",
|
||||
EventName::ObjectCreatedPut => "s3:ObjectCreated:Put",
|
||||
EventName::ObjectCreatedPutTagging => "s3:ObjectCreated:PutTagging",
|
||||
EventName::ObjectCreatedDeleteTagging => "s3:ObjectCreated:DeleteTagging",
|
||||
EventName::ObjectCreatedPutRetention => "s3:ObjectCreated:PutRetention",
|
||||
EventName::ObjectCreatedPutLegalHold => "s3:ObjectCreated:PutLegalHold",
|
||||
EventName::ObjectRemovedAll => "s3:ObjectRemoved:*",
|
||||
EventName::ObjectRemovedDelete => "s3:ObjectRemoved:Delete",
|
||||
EventName::ObjectRemovedDeleteMarkerCreated => "s3:ObjectRemoved:DeleteMarkerCreated",
|
||||
EventName::ObjectRemovedNoOP => "s3:ObjectRemoved:NoOP",
|
||||
EventName::ObjectRemovedDeleteAllVersions => "s3:ObjectRemoved:DeleteAllVersions",
|
||||
EventName::LifecycleDelMarkerExpirationDelete => "s3:LifecycleDelMarkerExpiration:Delete",
|
||||
EventName::ObjectReplicationAll => "s3:Replication:*",
|
||||
EventName::ObjectReplicationFailed => "s3:Replication:OperationFailedReplication",
|
||||
EventName::ObjectReplicationComplete => "s3:Replication:OperationCompletedReplication",
|
||||
EventName::ObjectReplicationNotTracked => "s3:Replication:OperationNotTracked",
|
||||
EventName::ObjectReplicationMissedThreshold => "s3:Replication:OperationMissedThreshold",
|
||||
EventName::ObjectReplicationReplicatedAfterThreshold => "s3:Replication:OperationReplicatedAfterThreshold",
|
||||
EventName::ObjectRestoreAll => "s3:ObjectRestore:*",
|
||||
EventName::ObjectRestorePost => "s3:ObjectRestore:Post",
|
||||
EventName::ObjectRestoreCompleted => "s3:ObjectRestore:Completed",
|
||||
EventName::ObjectTransitionAll => "s3:ObjectTransition:*",
|
||||
EventName::ObjectTransitionFailed => "s3:ObjectTransition:Failed",
|
||||
EventName::ObjectTransitionComplete => "s3:ObjectTransition:Complete",
|
||||
EventName::ScannerManyVersions => "s3:Scanner:ManyVersions",
|
||||
EventName::ScannerLargeVersions => "s3:Scanner:LargeVersions",
|
||||
EventName::ScannerBigPrefix => "s3:Scanner:BigPrefix",
|
||||
// Go's String() returns "" for ObjectScannerAll and Everything
|
||||
EventName::ObjectScannerAll => "s3:Scanner:*", // Follow the pattern in Go Expand
|
||||
EventName::Everything => "", // Go String() returns "" to unprocessed
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the extended value of the abbreviation event type.
|
||||
pub fn expand(&self) -> Vec<Self> {
|
||||
match self {
|
||||
EventName::ObjectAccessedAll => vec![
|
||||
EventName::ObjectAccessedGet,
|
||||
EventName::ObjectAccessedHead,
|
||||
EventName::ObjectAccessedGetRetention,
|
||||
EventName::ObjectAccessedGetLegalHold,
|
||||
EventName::ObjectAccessedAttributes,
|
||||
],
|
||||
EventName::ObjectCreatedAll => vec![
|
||||
EventName::ObjectCreatedCompleteMultipartUpload,
|
||||
EventName::ObjectCreatedCopy,
|
||||
EventName::ObjectCreatedPost,
|
||||
EventName::ObjectCreatedPut,
|
||||
EventName::ObjectCreatedPutRetention,
|
||||
EventName::ObjectCreatedPutLegalHold,
|
||||
EventName::ObjectCreatedPutTagging,
|
||||
EventName::ObjectCreatedDeleteTagging,
|
||||
],
|
||||
EventName::ObjectRemovedAll => vec![
|
||||
EventName::ObjectRemovedDelete,
|
||||
EventName::ObjectRemovedDeleteMarkerCreated,
|
||||
EventName::ObjectRemovedNoOP,
|
||||
EventName::ObjectRemovedDeleteAllVersions,
|
||||
],
|
||||
EventName::ObjectReplicationAll => vec![
|
||||
EventName::ObjectReplicationFailed,
|
||||
EventName::ObjectReplicationComplete,
|
||||
EventName::ObjectReplicationNotTracked,
|
||||
EventName::ObjectReplicationMissedThreshold,
|
||||
EventName::ObjectReplicationReplicatedAfterThreshold,
|
||||
],
|
||||
EventName::ObjectRestoreAll => vec![EventName::ObjectRestorePost, EventName::ObjectRestoreCompleted],
|
||||
EventName::ObjectTransitionAll => vec![EventName::ObjectTransitionFailed, EventName::ObjectTransitionComplete],
|
||||
EventName::ObjectScannerAll => vec![
|
||||
// New
|
||||
EventName::ScannerManyVersions,
|
||||
EventName::ScannerLargeVersions,
|
||||
EventName::ScannerBigPrefix,
|
||||
],
|
||||
EventName::Everything => {
|
||||
// New
|
||||
SINGLE_EVENT_NAMES_IN_ORDER.to_vec()
|
||||
}
|
||||
// A single type returns to itself directly
|
||||
_ => vec![*self],
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the mask of type.
|
||||
/// The compound "All" type will be expanded.
|
||||
pub fn mask(&self) -> u64 {
|
||||
let value = *self as u32;
|
||||
if value > 0 && value <= LAST_SINGLE_TYPE_VALUE {
|
||||
// It's a single type
|
||||
1u64 << (value - 1)
|
||||
} else {
|
||||
// It's a compound type
|
||||
let mut mask = 0u64;
|
||||
for n in self.expand() {
|
||||
mask |= n.mask(); // Recursively call mask
|
||||
}
|
||||
mask
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Display for EventName {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
write!(f, "{}", self.as_str())
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert to `EventName` according to string
|
||||
impl From<&str> for EventName {
|
||||
fn from(event_str: &str) -> Self {
|
||||
EventName::parse(event_str).unwrap_or_else(|e| panic!("{}", e))
|
||||
}
|
||||
}
|
||||
|
||||
/// Represents the identity of the user who triggered the event
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct Identity {
|
||||
@@ -532,17 +258,6 @@ fn initialize_response_elements(elements: &mut HashMap<String, String>, keys: &[
|
||||
}
|
||||
}
|
||||
|
||||
/// Represents a log of events for sending to targets
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct EventLog {
|
||||
/// The event name
|
||||
pub event_name: EventName,
|
||||
/// The object key
|
||||
pub key: String,
|
||||
/// The list of events
|
||||
pub records: Vec<Event>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct EventArgs {
|
||||
pub event_name: EventName,
|
||||
|
||||
@@ -12,19 +12,22 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::{
|
||||
error::TargetError,
|
||||
target::{Target, mqtt::MQTTArgs, webhook::WebhookArgs},
|
||||
};
|
||||
use async_trait::async_trait;
|
||||
use rumqttc::QoS;
|
||||
use rustfs_config::notify::{
|
||||
DEFAULT_DIR, DEFAULT_LIMIT, ENV_NOTIFY_MQTT_KEYS, ENV_NOTIFY_WEBHOOK_KEYS, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL,
|
||||
MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME,
|
||||
NOTIFY_MQTT_KEYS, NOTIFY_WEBHOOK_KEYS, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT,
|
||||
WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
|
||||
ENV_NOTIFY_MQTT_KEYS, ENV_NOTIFY_WEBHOOK_KEYS, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS,
|
||||
MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, NOTIFY_MQTT_KEYS, NOTIFY_WEBHOOK_KEYS,
|
||||
WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
|
||||
};
|
||||
|
||||
use crate::Event;
|
||||
use rustfs_config::{DEFAULT_DIR, DEFAULT_LIMIT};
|
||||
use rustfs_ecstore::config::KVS;
|
||||
use rustfs_targets::{
|
||||
Target,
|
||||
error::TargetError,
|
||||
target::{mqtt::MQTTArgs, webhook::WebhookArgs},
|
||||
};
|
||||
use std::collections::HashSet;
|
||||
use std::time::Duration;
|
||||
use tracing::{debug, warn};
|
||||
@@ -34,7 +37,7 @@ use url::Url;
|
||||
#[async_trait]
|
||||
pub trait TargetFactory: Send + Sync {
|
||||
/// Creates a target from configuration
|
||||
async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target + Send + Sync>, TargetError>;
|
||||
async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target<Event> + Send + Sync>, TargetError>;
|
||||
|
||||
/// Validates target configuration
|
||||
fn validate_config(&self, id: &str, config: &KVS) -> Result<(), TargetError>;
|
||||
@@ -53,7 +56,7 @@ pub struct WebhookTargetFactory;
|
||||
|
||||
#[async_trait]
|
||||
impl TargetFactory for WebhookTargetFactory {
|
||||
async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target + Send + Sync>, TargetError> {
|
||||
async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target<Event> + Send + Sync>, TargetError> {
|
||||
// All config values are now read directly from the merged `config` KVS.
|
||||
let endpoint = config
|
||||
.lookup(WEBHOOK_ENDPOINT)
|
||||
@@ -72,9 +75,10 @@ impl TargetFactory for WebhookTargetFactory {
|
||||
.unwrap_or(DEFAULT_LIMIT),
|
||||
client_cert: config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default(),
|
||||
client_key: config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default(),
|
||||
target_type: rustfs_targets::target::TargetType::NotifyEvent,
|
||||
};
|
||||
|
||||
let target = crate::target::webhook::WebhookTarget::new(id, args)?;
|
||||
let target = rustfs_targets::target::webhook::WebhookTarget::new(id, args)?;
|
||||
Ok(Box::new(target))
|
||||
}
|
||||
|
||||
@@ -119,7 +123,7 @@ pub struct MQTTTargetFactory;
|
||||
|
||||
#[async_trait]
|
||||
impl TargetFactory for MQTTTargetFactory {
|
||||
async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target + Send + Sync>, TargetError> {
|
||||
async fn create_target(&self, id: String, config: &KVS) -> Result<Box<dyn Target<Event> + Send + Sync>, TargetError> {
|
||||
let broker = config
|
||||
.lookup(MQTT_BROKER)
|
||||
.ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
|
||||
@@ -161,9 +165,10 @@ impl TargetFactory for MQTTTargetFactory {
|
||||
.lookup(MQTT_QUEUE_LIMIT)
|
||||
.and_then(|v| v.parse::<u64>().ok())
|
||||
.unwrap_or(DEFAULT_LIMIT),
|
||||
target_type: rustfs_targets::target::TargetType::NotifyEvent,
|
||||
};
|
||||
|
||||
let target = crate::target::mqtt::MQTTTarget::new(id, args)?;
|
||||
let target = rustfs_targets::target::mqtt::MQTTTarget::new(id, args)?;
|
||||
Ok(Box::new(target))
|
||||
}
|
||||
|
||||
|
||||
+106
-3
@@ -12,11 +12,13 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::{Event, EventArgs, NotificationError, NotificationSystem};
|
||||
use crate::{BucketNotificationConfig, Event, EventArgs, NotificationError, NotificationSystem};
|
||||
use once_cell::sync::Lazy;
|
||||
use rustfs_ecstore::config::Config;
|
||||
use rustfs_targets::EventName;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use std::sync::{Arc, OnceLock};
|
||||
use tracing::instrument;
|
||||
use tracing::{error, instrument};
|
||||
|
||||
static NOTIFICATION_SYSTEM: OnceLock<Arc<NotificationSystem>> = OnceLock::new();
|
||||
// Create a globally unique Notifier instance
|
||||
@@ -57,6 +59,14 @@ pub struct Notifier {}
|
||||
impl Notifier {
|
||||
/// Notify an event asynchronously.
|
||||
/// This is the only entry point for all event notifications in the system.
|
||||
/// # Parameter
|
||||
/// - `args`: The event arguments containing details about the event to be notified.
|
||||
///
|
||||
/// # Return value
|
||||
/// Returns `()`, indicating that the notification has been sent.
|
||||
///
|
||||
/// # Using
|
||||
/// This function is used to notify events in the system, such as object creation, deletion, or updates.
|
||||
#[instrument(skip(self, args))]
|
||||
pub async fn notify(&self, args: EventArgs) {
|
||||
// Dependency injection or service positioning mode obtain NotificationSystem instance
|
||||
@@ -64,7 +74,7 @@ impl Notifier {
|
||||
// If the notification system itself cannot be retrieved, it will be returned directly
|
||||
Some(sys) => sys,
|
||||
None => {
|
||||
tracing::error!("Notification system is not initialized.");
|
||||
error!("Notification system is not initialized.");
|
||||
return;
|
||||
}
|
||||
};
|
||||
@@ -76,6 +86,7 @@ impl Notifier {
|
||||
|
||||
// Check if any subscribers are interested in the event
|
||||
if !notification_sys.has_subscriber(&args.bucket_name, &args.event_name).await {
|
||||
error!("No subscribers for event: {} in bucket: {}", args.event_name, args.bucket_name);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -83,4 +94,96 @@ impl Notifier {
|
||||
let event = Arc::new(Event::new(args));
|
||||
notification_sys.send_event(event).await;
|
||||
}
|
||||
|
||||
/// Add notification rules for the specified bucket and load configuration
|
||||
/// # Parameter
|
||||
/// - `bucket_name`: The name of the target bucket.
|
||||
/// - `region`: The area where bucket is located.
|
||||
/// - `event_names`: A list of event names that trigger notifications.
|
||||
/// - `prefix`: The prefix of the object key that triggers notifications.
|
||||
/// - `suffix`: The suffix of the object key that triggers notifications.
|
||||
/// - `target_ids`: A list of target IDs that will receive notifications.
|
||||
///
|
||||
/// # Return value
|
||||
/// Returns `Result<(), NotificationError>`, Ok on success, and an error on failure
|
||||
///
|
||||
/// # Using
|
||||
/// This function allows you to dynamically add notification rules for a specific bucket.
|
||||
pub async fn add_bucket_notification_rule(
|
||||
&self,
|
||||
bucket_name: &str,
|
||||
region: &str,
|
||||
event_names: &[EventName],
|
||||
prefix: &str,
|
||||
suffix: &str,
|
||||
target_ids: &[TargetID],
|
||||
) -> Result<(), NotificationError> {
|
||||
// Construct pattern, simple splicing of prefixes and suffixes
|
||||
let mut pattern = String::new();
|
||||
if !prefix.is_empty() {
|
||||
pattern.push_str(prefix);
|
||||
}
|
||||
pattern.push('*');
|
||||
if !suffix.is_empty() {
|
||||
pattern.push_str(suffix);
|
||||
}
|
||||
|
||||
// Create BucketNotificationConfig
|
||||
let mut bucket_config = BucketNotificationConfig::new(region);
|
||||
for target_id in target_ids {
|
||||
bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
|
||||
}
|
||||
|
||||
// Get global NotificationSystem
|
||||
let notification_sys = match notification_system() {
|
||||
Some(sys) => sys,
|
||||
None => return Err(NotificationError::ServerNotInitialized),
|
||||
};
|
||||
|
||||
// Loading configuration
|
||||
notification_sys
|
||||
.load_bucket_notification_config(bucket_name, &bucket_config)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Dynamically add notification rules according to different event types.
|
||||
///
|
||||
/// # Parameter
|
||||
/// - `bucket_name`: The name of the target bucket.
|
||||
/// - `region`: The area where bucket is located.
|
||||
/// - `event_rules`: Each rule contains a list of event types, prefixes, suffixes, and target IDs.
|
||||
///
|
||||
/// # Return value
|
||||
/// Returns `Result<(), NotificationError>`, Ok on success, and an error on failure.
|
||||
///
|
||||
/// # Using
|
||||
/// Supports notification rules for adding multiple event types, prefixes, suffixes, and targets to the same bucket in batches.
|
||||
pub async fn add_event_specific_rules(
|
||||
&self,
|
||||
bucket_name: &str,
|
||||
region: &str,
|
||||
event_rules: &[(Vec<EventName>, &str, &str, Vec<TargetID>)],
|
||||
) -> Result<(), NotificationError> {
|
||||
let mut bucket_config = BucketNotificationConfig::new(region);
|
||||
|
||||
for (event_names, prefix, suffix, target_ids) in event_rules {
|
||||
// Use `new_pattern` to construct a matching pattern
|
||||
let pattern = crate::rules::pattern::new_pattern(Some(prefix), Some(suffix));
|
||||
|
||||
for target_id in target_ids {
|
||||
bucket_config.add_rule(event_names, pattern.clone(), target_id.clone());
|
||||
}
|
||||
}
|
||||
|
||||
// Get global NotificationSystem instance
|
||||
let notification_sys = match notification_system() {
|
||||
Some(sys) => sys,
|
||||
None => return Err(NotificationError::ServerNotInitialized),
|
||||
};
|
||||
|
||||
// Loading configuration
|
||||
notification_sys
|
||||
.load_bucket_notification_config(bucket_name, &bucket_config)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,13 +12,15 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::arn::TargetID;
|
||||
use crate::store::{Key, Store};
|
||||
use crate::{
|
||||
Event, EventName, StoreError, Target, error::NotificationError, notifier::EventNotifier, registry::TargetRegistry,
|
||||
rules::BucketNotificationConfig, stream,
|
||||
Event, error::NotificationError, notifier::EventNotifier, registry::TargetRegistry, rules::BucketNotificationConfig, stream,
|
||||
};
|
||||
use rustfs_ecstore::config::{Config, KVS};
|
||||
use rustfs_targets::EventName;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use rustfs_targets::store::{Key, Store};
|
||||
use rustfs_targets::target::EntityTarget;
|
||||
use rustfs_targets::{StoreError, Target};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
@@ -127,7 +129,7 @@ impl NotificationSystem {
|
||||
|
||||
let config = self.config.read().await;
|
||||
debug!("Initializing notification system with config: {:?}", *config);
|
||||
let targets: Vec<Box<dyn Target + Send + Sync>> = self.registry.create_targets_from_config(&config).await?;
|
||||
let targets: Vec<Box<dyn Target<Event> + Send + Sync>> = self.registry.create_targets_from_config(&config).await?;
|
||||
|
||||
info!("{} notification targets were created", targets.len());
|
||||
|
||||
@@ -318,8 +320,8 @@ impl NotificationSystem {
|
||||
/// Enhanced event stream startup function, including monitoring and concurrency control
|
||||
fn enhanced_start_event_stream(
|
||||
&self,
|
||||
store: Box<dyn Store<Event, Error = StoreError, Key = Key> + Send>,
|
||||
target: Arc<dyn Target + Send + Sync>,
|
||||
store: Box<dyn Store<EntityTarget<Event>, Error = StoreError, Key = Key> + Send>,
|
||||
target: Arc<dyn Target<Event> + Send + Sync>,
|
||||
metrics: Arc<NotificationMetrics>,
|
||||
semaphore: Arc<Semaphore>,
|
||||
) -> mpsc::Sender<()> {
|
||||
@@ -348,7 +350,7 @@ impl NotificationSystem {
|
||||
|
||||
// Create a new target from configuration
|
||||
// This function will now be responsible for merging env, creating and persisting the final configuration.
|
||||
let targets: Vec<Box<dyn Target + Send + Sync>> = self
|
||||
let targets: Vec<Box<dyn Target<Event> + Send + Sync>> = self
|
||||
.registry
|
||||
.create_targets_from_config(&new_config)
|
||||
.await
|
||||
|
||||
@@ -18,7 +18,6 @@
|
||||
//! It supports sending events to various targets
|
||||
//! (like Webhook and MQTT) and includes features like event persistence and retry on failure.
|
||||
|
||||
pub mod arn;
|
||||
pub mod error;
|
||||
pub mod event;
|
||||
pub mod factory;
|
||||
@@ -27,59 +26,10 @@ pub mod integration;
|
||||
pub mod notifier;
|
||||
pub mod registry;
|
||||
pub mod rules;
|
||||
pub mod store;
|
||||
pub mod stream;
|
||||
pub mod target;
|
||||
|
||||
// Re-exports
|
||||
pub use error::{NotificationError, StoreError, TargetError};
|
||||
pub use event::{Event, EventArgs, EventLog, EventName};
|
||||
pub use error::NotificationError;
|
||||
pub use event::{Event, EventArgs};
|
||||
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};
|
||||
|
||||
/// Initialize the tracing log system
|
||||
///
|
||||
/// # Example
|
||||
/// ```
|
||||
/// rustfs_notify::init_logger(rustfs_notify::LogLevel::Info);
|
||||
/// ```
|
||||
pub fn init_logger(level: LogLevel) {
|
||||
let filter = EnvFilter::default().add_directive(level.into());
|
||||
tracing_subscriber::registry()
|
||||
.with(filter)
|
||||
.with(
|
||||
fmt::layer()
|
||||
.with_target(true)
|
||||
.with_target(true)
|
||||
.with_ansi(std::io::stdout().is_terminal())
|
||||
.with_thread_names(true)
|
||||
.with_thread_ids(true)
|
||||
.with_file(true)
|
||||
.with_line_number(true),
|
||||
)
|
||||
.init();
|
||||
}
|
||||
|
||||
/// Log level definition
|
||||
pub enum LogLevel {
|
||||
Debug,
|
||||
Info,
|
||||
Warn,
|
||||
Error,
|
||||
}
|
||||
|
||||
impl From<LogLevel> for tracing_subscriber::filter::Directive {
|
||||
fn from(level: LogLevel) -> Self {
|
||||
match level {
|
||||
LogLevel::Debug => "debug".parse().unwrap(),
|
||||
LogLevel::Info => "info".parse().unwrap(),
|
||||
LogLevel::Warn => "warn".parse().unwrap(),
|
||||
LogLevel::Error => "error".parse().unwrap(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,9 +12,12 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::arn::TargetID;
|
||||
use crate::{EventName, error::NotificationError, event::Event, rules::RulesMap, target::Target};
|
||||
use crate::{error::NotificationError, event::Event, rules::RulesMap};
|
||||
use dashmap::DashMap;
|
||||
use rustfs_targets::EventName;
|
||||
use rustfs_targets::Target;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use rustfs_targets::target::EntityTarget;
|
||||
use std::{collections::HashMap, sync::Arc};
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, error, info, instrument, warn};
|
||||
@@ -121,7 +124,7 @@ impl EventNotifier {
|
||||
}
|
||||
|
||||
/// Sends an event to the appropriate targets based on the bucket rules
|
||||
#[instrument(skip(self, event))]
|
||||
#[instrument(skip_all)]
|
||||
pub async fn send(&self, event: Arc<Event>) {
|
||||
let bucket_name = &event.s3.bucket.name;
|
||||
let object_key = &event.s3.object.key;
|
||||
@@ -149,8 +152,15 @@ impl EventNotifier {
|
||||
let target_name_for_task = cloned_target_for_task.name(); // Get the name before generating the task
|
||||
debug!("Preparing to send event to target: {}", target_name_for_task);
|
||||
// Use cloned data in closures to avoid borrowing conflicts
|
||||
// Create an EntityTarget from the event
|
||||
let entity_target: Arc<EntityTarget<Event>> = Arc::new(EntityTarget {
|
||||
object_name: object_key.to_string(),
|
||||
bucket_name: bucket_name.to_string(),
|
||||
event_name,
|
||||
data: event_clone.clone().as_ref().clone(),
|
||||
});
|
||||
let handle = tokio::spawn(async move {
|
||||
if let Err(e) = cloned_target_for_task.save(event_clone).await {
|
||||
if let Err(e) = cloned_target_for_task.save(entity_target.clone()).await {
|
||||
error!("Failed to send event to target {}: {}", target_name_for_task, e);
|
||||
} else {
|
||||
debug!("Successfully saved event to target {}", target_name_for_task);
|
||||
@@ -180,7 +190,7 @@ impl EventNotifier {
|
||||
#[instrument(skip(self, targets_to_init))]
|
||||
pub async fn init_bucket_targets(
|
||||
&self,
|
||||
targets_to_init: Vec<Box<dyn Target + Send + Sync>>,
|
||||
targets_to_init: Vec<Box<dyn Target<Event> + Send + Sync>>,
|
||||
) -> Result<(), NotificationError> {
|
||||
// Currently active, simpler logic
|
||||
let mut target_list_guard = self.target_list.write().await; //Gets a write lock for the TargetList
|
||||
@@ -189,7 +199,7 @@ impl EventNotifier {
|
||||
debug!("init bucket target: {}", target_boxed.name());
|
||||
// TargetList::add method expectations Arc<dyn Target + Send + Sync>
|
||||
// Therefore, you need to convert Box<dyn Target + Send + Sync> to Arc<dyn Target + Send + Sync>
|
||||
let target_arc: Arc<dyn Target + Send + Sync> = Arc::from(target_boxed);
|
||||
let target_arc: Arc<dyn Target<Event> + Send + Sync> = Arc::from(target_boxed);
|
||||
target_list_guard.add(target_arc)?; // Add Arc<dyn Target> to the list
|
||||
}
|
||||
info!(
|
||||
@@ -203,7 +213,7 @@ impl EventNotifier {
|
||||
|
||||
/// A thread-safe list of targets
|
||||
pub struct TargetList {
|
||||
targets: HashMap<TargetID, Arc<dyn Target + Send + Sync>>,
|
||||
targets: HashMap<TargetID, Arc<dyn Target<Event> + Send + Sync>>,
|
||||
}
|
||||
|
||||
impl Default for TargetList {
|
||||
@@ -219,7 +229,7 @@ impl TargetList {
|
||||
}
|
||||
|
||||
/// Adds a target to the list
|
||||
pub fn add(&mut self, target: Arc<dyn Target + Send + Sync>) -> Result<(), NotificationError> {
|
||||
pub fn add(&mut self, target: Arc<dyn Target<Event> + Send + Sync>) -> Result<(), NotificationError> {
|
||||
let id = target.id();
|
||||
if self.targets.contains_key(&id) {
|
||||
// Potentially update or log a warning/error if replacing an existing target.
|
||||
@@ -231,7 +241,7 @@ impl TargetList {
|
||||
|
||||
/// Removes a target by ID. Note: This does not stop its associated event stream.
|
||||
/// Stream cancellation should be handled by EventNotifier.
|
||||
pub async fn remove_target_only(&mut self, id: &TargetID) -> Option<Arc<dyn Target + Send + Sync>> {
|
||||
pub async fn remove_target_only(&mut self, id: &TargetID) -> Option<Arc<dyn Target<Event> + Send + Sync>> {
|
||||
if let Some(target_arc) = self.targets.remove(id) {
|
||||
if let Err(e) = target_arc.close().await {
|
||||
// Target's own close logic
|
||||
@@ -258,7 +268,7 @@ impl TargetList {
|
||||
}
|
||||
|
||||
/// Returns a target by ID
|
||||
pub fn get(&self, id: &TargetID) -> Option<Arc<dyn Target + Send + Sync>> {
|
||||
pub fn get(&self, id: &TargetID) -> Option<Arc<dyn Target<Event> + Send + Sync>> {
|
||||
self.targets.get(id).cloned()
|
||||
}
|
||||
|
||||
|
||||
@@ -12,16 +12,15 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::target::ChannelTargetType;
|
||||
use crate::{
|
||||
error::TargetError,
|
||||
factory::{MQTTTargetFactory, TargetFactory, WebhookTargetFactory},
|
||||
target::Target,
|
||||
};
|
||||
use crate::Event;
|
||||
use crate::factory::{MQTTTargetFactory, TargetFactory, WebhookTargetFactory};
|
||||
use futures::stream::{FuturesUnordered, StreamExt};
|
||||
use rustfs_config::notify::{ENABLE_KEY, NOTIFY_ROUTE_PREFIX};
|
||||
use rustfs_config::{DEFAULT_DELIMITER, ENV_PREFIX};
|
||||
use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
|
||||
use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX};
|
||||
use rustfs_ecstore::config::{Config, KVS};
|
||||
use rustfs_targets::Target;
|
||||
use rustfs_targets::TargetError;
|
||||
use rustfs_targets::target::ChannelTargetType;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
@@ -61,7 +60,7 @@ impl TargetRegistry {
|
||||
target_type: &str,
|
||||
id: String,
|
||||
config: &KVS,
|
||||
) -> Result<Box<dyn Target + Send + Sync>, TargetError> {
|
||||
) -> Result<Box<dyn Target<Event> + Send + Sync>, TargetError> {
|
||||
let factory = self
|
||||
.factories
|
||||
.get(target_type)
|
||||
@@ -83,7 +82,10 @@ impl TargetRegistry {
|
||||
/// 4. Combine the default configuration, file configuration, and environment variable configuration for each instance.
|
||||
/// 5. If the instance is enabled, create an asynchronous task for it to instantiate.
|
||||
/// 6. Concurrency executes all creation tasks and collects results.
|
||||
pub async fn create_targets_from_config(&self, config: &Config) -> Result<Vec<Box<dyn Target + Send + Sync>>, TargetError> {
|
||||
pub async fn create_targets_from_config(
|
||||
&self,
|
||||
config: &Config,
|
||||
) -> Result<Vec<Box<dyn Target<Event> + Send + Sync>>, TargetError> {
|
||||
// Collect only environment variables with the relevant prefix to reduce memory usage
|
||||
let all_env: Vec<(String, String)> = std::env::vars().filter(|(key, _)| key.starts_with(ENV_PREFIX)).collect();
|
||||
// A collection of asynchronous tasks for concurrently executing target creation
|
||||
|
||||
@@ -14,11 +14,11 @@
|
||||
|
||||
use super::rules_map::RulesMap;
|
||||
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 rustfs_targets::EventName;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
use std::io::Read;
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
use super::pattern;
|
||||
use super::target_id_set::TargetIdSet;
|
||||
use crate::arn::TargetID;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
|
||||
|
||||
@@ -14,8 +14,8 @@
|
||||
|
||||
use super::pattern_rules::PatternRules;
|
||||
use super::target_id_set::TargetIdSet;
|
||||
use crate::arn::TargetID;
|
||||
use crate::event::EventName;
|
||||
use rustfs_targets::EventName;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::arn::TargetID;
|
||||
use rustfs_targets::arn::TargetID;
|
||||
use std::collections::HashSet;
|
||||
|
||||
/// TargetIDSet - A collection representation of TargetID.
|
||||
|
||||
@@ -13,8 +13,8 @@
|
||||
// limitations under the License.
|
||||
|
||||
use super::pattern;
|
||||
use crate::arn::{ARN, ArnError, TargetIDError};
|
||||
use crate::event::EventName;
|
||||
use rustfs_targets::EventName;
|
||||
use rustfs_targets::arn::{ARN, ArnError, TargetIDError};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashSet;
|
||||
use std::io::Read;
|
||||
|
||||
@@ -1,490 +0,0 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::error::StoreError;
|
||||
use rustfs_config::notify::{COMPRESS_EXT, DEFAULT_EXT, DEFAULT_LIMIT};
|
||||
use serde::{Serialize, de::DeserializeOwned};
|
||||
use snap::raw::{Decoder, Encoder};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::{
|
||||
collections::HashMap,
|
||||
marker::PhantomData,
|
||||
path::PathBuf,
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
use tracing::{debug, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
/// Represents a key for an entry in the store
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Key {
|
||||
/// The name of the key (UUID)
|
||||
pub name: String,
|
||||
/// The file extension for the entry
|
||||
pub extension: String,
|
||||
/// The number of items in the entry (for batch storage)
|
||||
pub item_count: usize,
|
||||
/// Whether the entry is compressed
|
||||
pub compress: bool,
|
||||
}
|
||||
|
||||
impl Key {
|
||||
/// Converts the key to a string (filename)
|
||||
pub fn to_key_string(&self) -> String {
|
||||
let name_part = if self.item_count > 1 {
|
||||
format!("{}:{}", self.item_count, self.name)
|
||||
} else {
|
||||
self.name.clone()
|
||||
};
|
||||
|
||||
let mut file_name = name_part;
|
||||
if !self.extension.is_empty() {
|
||||
file_name.push_str(&self.extension);
|
||||
}
|
||||
|
||||
if self.compress {
|
||||
file_name.push_str(COMPRESS_EXT);
|
||||
}
|
||||
file_name
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for Key {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
let name_part = if self.item_count > 1 {
|
||||
format!("{}:{}", self.item_count, self.name)
|
||||
} else {
|
||||
self.name.clone()
|
||||
};
|
||||
|
||||
let mut file_name = name_part;
|
||||
if !self.extension.is_empty() {
|
||||
file_name.push_str(&self.extension);
|
||||
}
|
||||
|
||||
if self.compress {
|
||||
file_name.push_str(COMPRESS_EXT);
|
||||
}
|
||||
write!(f, "{file_name}")
|
||||
}
|
||||
}
|
||||
|
||||
/// Parses a string into a Key
|
||||
pub fn parse_key(s: &str) -> Key {
|
||||
debug!("Parsing key: {}", s);
|
||||
|
||||
let mut name = s.to_string();
|
||||
let mut extension = String::new();
|
||||
let mut item_count = 1;
|
||||
let mut compress = false;
|
||||
|
||||
// Check for compressed suffixes
|
||||
if name.ends_with(COMPRESS_EXT) {
|
||||
compress = true;
|
||||
name = name[..name.len() - COMPRESS_EXT.len()].to_string();
|
||||
}
|
||||
|
||||
// Number of batch items parsed
|
||||
if let Some(colon_pos) = name.find(':') {
|
||||
if let Ok(count) = name[..colon_pos].parse::<usize>() {
|
||||
item_count = count;
|
||||
name = name[colon_pos + 1..].to_string();
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve extension
|
||||
if let Some(dot_pos) = name.rfind('.') {
|
||||
extension = name[dot_pos..].to_string();
|
||||
name = name[..dot_pos].to_string();
|
||||
}
|
||||
|
||||
debug!(
|
||||
"Parsed key - name: {}, extension: {}, item_count: {}, compress: {}",
|
||||
name, extension, item_count, compress
|
||||
);
|
||||
|
||||
Key {
|
||||
name,
|
||||
extension,
|
||||
item_count,
|
||||
compress,
|
||||
}
|
||||
}
|
||||
|
||||
/// Trait for a store that can store and retrieve items of type T
|
||||
pub trait Store<T>: Send + Sync {
|
||||
/// The error type for the store
|
||||
type Error;
|
||||
/// The key type for the store
|
||||
type Key;
|
||||
|
||||
/// Opens the store
|
||||
fn open(&self) -> Result<(), Self::Error>;
|
||||
|
||||
/// Stores a single item
|
||||
fn put(&self, item: Arc<T>) -> Result<Self::Key, Self::Error>;
|
||||
|
||||
/// Stores multiple items in a single batch
|
||||
fn put_multiple(&self, items: Vec<T>) -> Result<Self::Key, Self::Error>;
|
||||
|
||||
/// Retrieves a single item by key
|
||||
fn get(&self, key: &Self::Key) -> Result<T, Self::Error>;
|
||||
|
||||
/// Retrieves multiple items by key
|
||||
fn get_multiple(&self, key: &Self::Key) -> Result<Vec<T>, Self::Error>;
|
||||
|
||||
/// Deletes an item by key
|
||||
fn del(&self, key: &Self::Key) -> Result<(), Self::Error>;
|
||||
|
||||
/// Lists all keys in the store
|
||||
fn list(&self) -> Vec<Self::Key>;
|
||||
|
||||
/// Returns the number of items in the store
|
||||
fn len(&self) -> usize;
|
||||
|
||||
/// Returns true if the store is empty
|
||||
fn is_empty(&self) -> bool;
|
||||
|
||||
/// Clones the store into a boxed trait object
|
||||
fn boxed_clone(&self) -> Box<dyn Store<T, Error = Self::Error, Key = Self::Key> + Send + Sync>;
|
||||
}
|
||||
|
||||
/// A store that uses the filesystem to persist events in a queue
|
||||
pub struct QueueStore<T> {
|
||||
entry_limit: u64,
|
||||
directory: PathBuf,
|
||||
file_ext: String,
|
||||
entries: Arc<RwLock<HashMap<String, i64>>>, // key -> modtime as unix nano
|
||||
_phantom: PhantomData<T>,
|
||||
}
|
||||
|
||||
impl<T> Clone for QueueStore<T> {
|
||||
fn clone(&self) -> Self {
|
||||
QueueStore {
|
||||
entry_limit: self.entry_limit,
|
||||
directory: self.directory.clone(),
|
||||
file_ext: self.file_ext.clone(),
|
||||
entries: Arc::clone(&self.entries),
|
||||
_phantom: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Serialize + DeserializeOwned + Send + Sync> QueueStore<T> {
|
||||
/// Creates a new QueueStore
|
||||
pub fn new(directory: impl Into<PathBuf>, limit: u64, ext: &str) -> Self {
|
||||
let file_ext = if ext.is_empty() { DEFAULT_EXT } else { ext };
|
||||
|
||||
QueueStore {
|
||||
directory: directory.into(),
|
||||
entry_limit: if limit == 0 { DEFAULT_LIMIT } else { limit },
|
||||
file_ext: file_ext.to_string(),
|
||||
entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))),
|
||||
_phantom: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the full path for a key
|
||||
fn file_path(&self, key: &Key) -> PathBuf {
|
||||
self.directory.join(key.to_string())
|
||||
}
|
||||
|
||||
/// Reads a file for the given key
|
||||
fn read_file(&self, key: &Key) -> Result<Vec<u8>, StoreError> {
|
||||
let path = self.file_path(key);
|
||||
debug!("Reading file for key: {},path: {}", key.to_string(), path.display());
|
||||
let data = std::fs::read(&path).map_err(|e| {
|
||||
if e.kind() == std::io::ErrorKind::NotFound {
|
||||
StoreError::NotFound
|
||||
} else {
|
||||
StoreError::Io(e)
|
||||
}
|
||||
})?;
|
||||
|
||||
if data.is_empty() {
|
||||
return Err(StoreError::NotFound);
|
||||
}
|
||||
|
||||
if key.compress {
|
||||
let mut decoder = Decoder::new();
|
||||
decoder
|
||||
.decompress_vec(&data)
|
||||
.map_err(|e| StoreError::Compression(e.to_string()))
|
||||
} else {
|
||||
Ok(data)
|
||||
}
|
||||
}
|
||||
|
||||
/// Writes data to a file for the given key
|
||||
fn write_file(&self, key: &Key, data: &[u8]) -> Result<(), StoreError> {
|
||||
let path = self.file_path(key);
|
||||
// Create directory if it doesn't exist
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(StoreError::Io)?;
|
||||
}
|
||||
|
||||
let data = if key.compress {
|
||||
let mut encoder = Encoder::new();
|
||||
encoder
|
||||
.compress_vec(data)
|
||||
.map_err(|e| StoreError::Compression(e.to_string()))?
|
||||
} else {
|
||||
data.to_vec()
|
||||
};
|
||||
|
||||
std::fs::write(&path, &data).map_err(StoreError::Io)?;
|
||||
let modified = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
|
||||
let mut entries = self
|
||||
.entries
|
||||
.write()
|
||||
.map_err(|_| StoreError::Internal("Failed to acquire write lock on entries".to_string()))?;
|
||||
entries.insert(key.to_string(), modified);
|
||||
debug!("Wrote event to store: {}", key.to_string());
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Store<T> for QueueStore<T>
|
||||
where
|
||||
T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
|
||||
{
|
||||
type Error = StoreError;
|
||||
type Key = Key;
|
||||
|
||||
fn open(&self) -> Result<(), Self::Error> {
|
||||
std::fs::create_dir_all(&self.directory).map_err(StoreError::Io)?;
|
||||
|
||||
let entries = std::fs::read_dir(&self.directory).map_err(StoreError::Io)?;
|
||||
// Get the write lock to update the internal state
|
||||
let mut entries_map = self
|
||||
.entries
|
||||
.write()
|
||||
.map_err(|_| StoreError::Internal("Failed to acquire write lock on entries".to_string()))?;
|
||||
for entry in entries {
|
||||
let entry = entry.map_err(StoreError::Io)?;
|
||||
let metadata = entry.metadata().map_err(StoreError::Io)?;
|
||||
if metadata.is_file() {
|
||||
let modified = metadata.modified().map_err(StoreError::Io)?;
|
||||
let unix_nano = modified.duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as i64;
|
||||
|
||||
let file_name = entry.file_name().to_string_lossy().to_string();
|
||||
entries_map.insert(file_name, unix_nano);
|
||||
}
|
||||
}
|
||||
|
||||
debug!("Opened store at: {:?}", self.directory);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn put(&self, item: Arc<T>) -> Result<Self::Key, Self::Error> {
|
||||
// Check storage limits
|
||||
{
|
||||
let entries = self
|
||||
.entries
|
||||
.read()
|
||||
.map_err(|_| StoreError::Internal("Failed to acquire read lock on entries".to_string()))?;
|
||||
|
||||
if entries.len() as u64 >= self.entry_limit {
|
||||
return Err(StoreError::LimitExceeded);
|
||||
}
|
||||
}
|
||||
|
||||
let uuid = Uuid::new_v4();
|
||||
let key = Key {
|
||||
name: uuid.to_string(),
|
||||
extension: self.file_ext.clone(),
|
||||
item_count: 1,
|
||||
compress: true,
|
||||
};
|
||||
|
||||
let data = serde_json::to_vec(&item).map_err(|e| StoreError::Serialization(e.to_string()))?;
|
||||
self.write_file(&key, &data)?;
|
||||
|
||||
Ok(key)
|
||||
}
|
||||
|
||||
fn put_multiple(&self, items: Vec<T>) -> Result<Self::Key, Self::Error> {
|
||||
// Check storage limits
|
||||
{
|
||||
let entries = self
|
||||
.entries
|
||||
.read()
|
||||
.map_err(|_| StoreError::Internal("Failed to acquire read lock on entries".to_string()))?;
|
||||
|
||||
if entries.len() as u64 >= self.entry_limit {
|
||||
return Err(StoreError::LimitExceeded);
|
||||
}
|
||||
}
|
||||
if items.is_empty() {
|
||||
// Or return an error, or a special key?
|
||||
return Err(StoreError::Internal("Cannot put_multiple with empty items list".to_string()));
|
||||
}
|
||||
let uuid = Uuid::new_v4();
|
||||
let key = Key {
|
||||
name: uuid.to_string(),
|
||||
extension: self.file_ext.clone(),
|
||||
item_count: items.len(),
|
||||
compress: true,
|
||||
};
|
||||
|
||||
// Serialize all items into a single Vec<u8>
|
||||
// This current approach for get_multiple/put_multiple assumes items are concatenated JSON objects.
|
||||
// This might be problematic for deserialization if not handled carefully.
|
||||
// A better approach for multiple items might be to store them as a JSON array `Vec<T>`.
|
||||
// For now, sticking to current logic of concatenating.
|
||||
let mut buffer = Vec::new();
|
||||
for item in items {
|
||||
// If items are Vec<Event>, and Event is large, this could be inefficient.
|
||||
// The current get_multiple deserializes one by one.
|
||||
let item_data = serde_json::to_vec(&item).map_err(|e| StoreError::Serialization(e.to_string()))?;
|
||||
buffer.extend_from_slice(&item_data);
|
||||
// If using JSON array: buffer = serde_json::to_vec(&items)?
|
||||
}
|
||||
|
||||
self.write_file(&key, &buffer)?;
|
||||
|
||||
Ok(key)
|
||||
}
|
||||
|
||||
fn get(&self, key: &Self::Key) -> Result<T, Self::Error> {
|
||||
if key.item_count != 1 {
|
||||
return Err(StoreError::Internal(format!(
|
||||
"get() called on a batch key ({} items), use get_multiple()",
|
||||
key.item_count
|
||||
)));
|
||||
}
|
||||
let items = self.get_multiple(key)?;
|
||||
items.into_iter().next().ok_or(StoreError::NotFound)
|
||||
}
|
||||
|
||||
fn get_multiple(&self, key: &Self::Key) -> Result<Vec<T>, Self::Error> {
|
||||
debug!("Reading items from store for key: {}", key.to_string());
|
||||
let data = self.read_file(key)?;
|
||||
if data.is_empty() {
|
||||
return Err(StoreError::Deserialization("Cannot deserialize empty data".to_string()));
|
||||
}
|
||||
let mut items = Vec::with_capacity(key.item_count);
|
||||
|
||||
// let mut deserializer = serde_json::Deserializer::from_slice(&data);
|
||||
// while let Ok(item) = serde::Deserialize::deserialize(&mut deserializer) {
|
||||
// items.push(item);
|
||||
// }
|
||||
|
||||
// This deserialization logic assumes multiple JSON objects are simply concatenated in the file.
|
||||
// This is fragile. It's better to store a JSON array `[item1, item2, ...]`
|
||||
// or use a streaming deserializer that can handle multiple top-level objects if that's the format.
|
||||
// For now, assuming serde_json::Deserializer::from_slice can handle this if input is well-formed.
|
||||
let mut deserializer = serde_json::Deserializer::from_slice(&data).into_iter::<T>();
|
||||
|
||||
for _ in 0..key.item_count {
|
||||
match deserializer.next() {
|
||||
Some(Ok(item)) => items.push(item),
|
||||
Some(Err(e)) => {
|
||||
return Err(StoreError::Deserialization(format!("Failed to deserialize item in batch: {e}")));
|
||||
}
|
||||
None => {
|
||||
// Reached end of stream sooner than item_count
|
||||
if items.len() < key.item_count && !items.is_empty() {
|
||||
// Partial read
|
||||
warn!(
|
||||
"Expected {} items for key {}, but only found {}. Possible data corruption or incorrect item_count.",
|
||||
key.item_count,
|
||||
key.to_string(),
|
||||
items.len()
|
||||
);
|
||||
// Depending on strictness, this could be an error.
|
||||
} else if items.is_empty() {
|
||||
// No items at all, but file existed
|
||||
return Err(StoreError::Deserialization(format!(
|
||||
"No items deserialized for key {key} though file existed."
|
||||
)));
|
||||
}
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if items.is_empty() && key.item_count > 0 {
|
||||
return Err(StoreError::Deserialization("No items found".to_string()));
|
||||
}
|
||||
|
||||
Ok(items)
|
||||
}
|
||||
|
||||
fn del(&self, key: &Self::Key) -> Result<(), Self::Error> {
|
||||
let path = self.file_path(key);
|
||||
std::fs::remove_file(&path).map_err(|e| {
|
||||
if e.kind() == std::io::ErrorKind::NotFound {
|
||||
// If file not found, still try to remove from entries map in case of inconsistency
|
||||
warn!(
|
||||
"File not found for key {} during del, but proceeding to remove from entries map.",
|
||||
key.to_string()
|
||||
);
|
||||
StoreError::NotFound
|
||||
} else {
|
||||
StoreError::Io(e)
|
||||
}
|
||||
})?;
|
||||
|
||||
// Get the write lock to update the internal state
|
||||
let mut entries = self
|
||||
.entries
|
||||
.write()
|
||||
.map_err(|_| StoreError::Internal("Failed to acquire write lock on entries".to_string()))?;
|
||||
|
||||
if entries.remove(&key.to_string()).is_none() {
|
||||
// Key was not in the map, could be an inconsistency or already deleted.
|
||||
// This is not necessarily an error if the file deletion succeeded or was NotFound.
|
||||
debug!("Key {} not found in entries map during del, might have been already removed.", key);
|
||||
}
|
||||
debug!("Deleted event from store: {}", key.to_string());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn list(&self) -> Vec<Self::Key> {
|
||||
// Get the read lock to read the internal state
|
||||
let entries = match self.entries.read() {
|
||||
Ok(entries) => entries,
|
||||
Err(_) => {
|
||||
debug!("Failed to acquire read lock on entries for listing");
|
||||
return Vec::new();
|
||||
}
|
||||
};
|
||||
|
||||
let mut entries_vec: Vec<_> = entries.iter().collect();
|
||||
// Sort by modtime (value in HashMap) to process oldest first
|
||||
entries_vec.sort_by(|a, b| a.1.cmp(b.1)); // Oldest first
|
||||
|
||||
entries_vec.into_iter().map(|(k, _)| parse_key(k)).collect()
|
||||
}
|
||||
|
||||
fn len(&self) -> usize {
|
||||
// Get the read lock to read the internal state
|
||||
match self.entries.read() {
|
||||
Ok(entries) => entries.len(),
|
||||
Err(_) => {
|
||||
debug!("Failed to acquire read lock on entries for len");
|
||||
0
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn is_empty(&self) -> bool {
|
||||
self.len() == 0
|
||||
}
|
||||
|
||||
fn boxed_clone(&self) -> Box<dyn Store<T, Error = Self::Error, Key = Self::Key> + Send + Sync> {
|
||||
Box::new(self.clone()) as Box<dyn Store<T, Error = Self::Error, Key = Self::Key> + Send + Sync>
|
||||
}
|
||||
}
|
||||
+15
-16
@@ -12,13 +12,12 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::{
|
||||
Event, StoreError,
|
||||
error::TargetError,
|
||||
integration::NotificationMetrics,
|
||||
store::{Key, Store},
|
||||
target::Target,
|
||||
};
|
||||
use crate::{Event, integration::NotificationMetrics};
|
||||
use rustfs_targets::StoreError;
|
||||
use rustfs_targets::Target;
|
||||
use rustfs_targets::TargetError;
|
||||
use rustfs_targets::store::{Key, Store};
|
||||
use rustfs_targets::target::EntityTarget;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::sync::{Semaphore, mpsc};
|
||||
@@ -28,7 +27,7 @@ use tracing::{debug, error, info, warn};
|
||||
/// Streams events from the store to the target
|
||||
pub async fn stream_events(
|
||||
store: &mut (dyn Store<Event, Error = StoreError, Key = Key> + Send),
|
||||
target: &dyn Target,
|
||||
target: &dyn Target<Event>,
|
||||
mut cancel_rx: mpsc::Receiver<()>,
|
||||
) {
|
||||
info!("Starting event stream for target: {}", target.name());
|
||||
@@ -107,7 +106,7 @@ pub async fn stream_events(
|
||||
/// Starts the event streaming process for a target
|
||||
pub fn start_event_stream(
|
||||
mut store: Box<dyn Store<Event, Error = StoreError, Key = Key> + Send>,
|
||||
target: Arc<dyn Target + Send + Sync>,
|
||||
target: Arc<dyn Target<Event> + Send + Sync>,
|
||||
) -> mpsc::Sender<()> {
|
||||
let (cancel_tx, cancel_rx) = mpsc::channel(1);
|
||||
|
||||
@@ -121,8 +120,8 @@ pub fn start_event_stream(
|
||||
|
||||
/// Start event stream with batch processing
|
||||
pub fn start_event_stream_with_batching(
|
||||
mut store: Box<dyn Store<Event, Error = StoreError, Key = Key> + Send>,
|
||||
target: Arc<dyn Target + Send + Sync>,
|
||||
mut store: Box<dyn Store<EntityTarget<Event>, Error = StoreError, Key = Key> + Send>,
|
||||
target: Arc<dyn Target<Event> + Send + Sync>,
|
||||
metrics: Arc<NotificationMetrics>,
|
||||
semaphore: Arc<Semaphore>,
|
||||
) -> mpsc::Sender<()> {
|
||||
@@ -138,8 +137,8 @@ pub fn start_event_stream_with_batching(
|
||||
|
||||
/// Event stream processing with batch processing
|
||||
pub async fn stream_events_with_batching(
|
||||
store: &mut (dyn Store<Event, Error = StoreError, Key = Key> + Send),
|
||||
target: &dyn Target,
|
||||
store: &mut (dyn Store<EntityTarget<Event>, Error = StoreError, Key = Key> + Send),
|
||||
target: &dyn Target<Event>,
|
||||
mut cancel_rx: mpsc::Receiver<()>,
|
||||
metrics: Arc<NotificationMetrics>,
|
||||
semaphore: Arc<Semaphore>,
|
||||
@@ -156,7 +155,7 @@ pub async fn stream_events_with_batching(
|
||||
const MAX_RETRIES: usize = 5;
|
||||
const BASE_RETRY_DELAY: Duration = Duration::from_secs(2);
|
||||
|
||||
let mut batch = Vec::with_capacity(batch_size);
|
||||
let mut batch: Vec<EntityTarget<Event>> = Vec::with_capacity(batch_size);
|
||||
let mut batch_keys = Vec::with_capacity(batch_size);
|
||||
let mut last_flush = Instant::now();
|
||||
|
||||
@@ -234,9 +233,9 @@ pub async fn stream_events_with_batching(
|
||||
|
||||
/// Processing event batches
|
||||
async fn process_batch(
|
||||
batch: &mut Vec<Event>,
|
||||
batch: &mut Vec<EntityTarget<Event>>,
|
||||
batch_keys: &mut Vec<Key>,
|
||||
target: &dyn Target,
|
||||
target: &dyn Target<Event>,
|
||||
max_retries: usize,
|
||||
base_delay: Duration,
|
||||
metrics: &Arc<NotificationMetrics>,
|
||||
|
||||
@@ -1,119 +0,0 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
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;
|
||||
|
||||
/// Trait for notification targets
|
||||
#[async_trait]
|
||||
pub trait Target: Send + Sync + 'static {
|
||||
/// Returns the ID of the target
|
||||
fn id(&self) -> TargetID;
|
||||
|
||||
/// Returns the name of the target
|
||||
fn name(&self) -> String {
|
||||
self.id().to_string()
|
||||
}
|
||||
|
||||
/// Checks if the target is active and reachable
|
||||
async fn is_active(&self) -> Result<bool, TargetError>;
|
||||
|
||||
/// Saves an event (either sends it immediately or stores it for later)
|
||||
async fn save(&self, event: Arc<Event>) -> Result<(), TargetError>;
|
||||
|
||||
/// Sends an event from the store
|
||||
async fn send_from_store(&self, key: Key) -> Result<(), TargetError>;
|
||||
|
||||
/// Closes the target and releases resources
|
||||
async fn close(&self) -> Result<(), TargetError>;
|
||||
|
||||
/// Returns the store associated with the target (if any)
|
||||
fn store(&self) -> Option<&(dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync)>;
|
||||
|
||||
/// Returns the type of the target
|
||||
fn clone_dyn(&self) -> Box<dyn Target + Send + Sync>;
|
||||
|
||||
/// Initialize the target, such as establishing a connection, etc.
|
||||
async fn init(&self) -> Result<(), TargetError> {
|
||||
// The default implementation is empty
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Check if the target is enabled
|
||||
fn is_enabled(&self) -> bool;
|
||||
}
|
||||
|
||||
/// The `ChannelTargetType` enum represents the different types of channel Target
|
||||
/// used in the notification system.
|
||||
///
|
||||
/// It includes:
|
||||
/// - `Webhook`: Represents a webhook target for sending notifications via HTTP requests.
|
||||
/// - `Kafka`: Represents a Kafka target for sending notifications to a Kafka topic.
|
||||
/// - `Mqtt`: Represents an MQTT target for sending notifications via MQTT protocol.
|
||||
///
|
||||
/// Each variant has an associated string representation that can be used for serialization
|
||||
/// or logging purposes.
|
||||
/// The `as_str` method returns the string representation of the target type,
|
||||
/// and the `Display` implementation allows for easy formatting of the target type as a string.
|
||||
///
|
||||
/// example usage:
|
||||
/// ```rust
|
||||
/// use rustfs_notify::target::ChannelTargetType;
|
||||
///
|
||||
/// let target_type = ChannelTargetType::Webhook;
|
||||
/// assert_eq!(target_type.as_str(), "webhook");
|
||||
/// println!("Target type: {}", target_type);
|
||||
/// ```
|
||||
///
|
||||
/// example output:
|
||||
/// Target type: webhook
|
||||
pub enum ChannelTargetType {
|
||||
Webhook,
|
||||
Kafka,
|
||||
Mqtt,
|
||||
}
|
||||
|
||||
impl ChannelTargetType {
|
||||
pub fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
ChannelTargetType::Webhook => "webhook",
|
||||
ChannelTargetType::Kafka => "kafka",
|
||||
ChannelTargetType::Mqtt => "mqtt",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::fmt::Display for ChannelTargetType {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
ChannelTargetType::Webhook => write!(f, "webhook"),
|
||||
ChannelTargetType::Kafka => write!(f, "kafka"),
|
||||
ChannelTargetType::Mqtt => write!(f, "mqtt"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn parse_bool(value: &str) -> Result<bool, TargetError> {
|
||||
match value.to_lowercase().as_str() {
|
||||
"true" | "on" | "yes" | "1" => Ok(true),
|
||||
"false" | "off" | "no" | "0" => Ok(false),
|
||||
_ => Err(TargetError::ParseError(format!("Unable to parse boolean: {value}"))),
|
||||
}
|
||||
}
|
||||
@@ -1,643 +0,0 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::store::Key;
|
||||
use crate::target::ChannelTargetType;
|
||||
use crate::{
|
||||
StoreError, Target,
|
||||
arn::TargetID,
|
||||
error::TargetError,
|
||||
event::{Event, EventLog},
|
||||
store::Store,
|
||||
};
|
||||
use async_trait::async_trait;
|
||||
use rumqttc::{AsyncClient, EventLoop, MqttOptions, Outgoing, Packet, QoS};
|
||||
use rumqttc::{ConnectionError, mqttbytes::Error as MqttBytesError};
|
||||
use rustfs_config::notify::STORE_EXTENSION;
|
||||
use std::sync::Arc;
|
||||
use std::{
|
||||
path::PathBuf,
|
||||
sync::atomic::{AtomicBool, Ordering},
|
||||
time::Duration,
|
||||
};
|
||||
use tokio::sync::{Mutex, OnceCell, mpsc};
|
||||
use tracing::{debug, error, info, instrument, trace, warn};
|
||||
use url::Url;
|
||||
use urlencoding;
|
||||
|
||||
const DEFAULT_CONNECTION_TIMEOUT: Duration = Duration::from_secs(15);
|
||||
const EVENT_LOOP_POLL_TIMEOUT: Duration = Duration::from_secs(10); // For initial connection check in task
|
||||
|
||||
/// Arguments for configuring an MQTT target
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MQTTArgs {
|
||||
/// Whether the target is enabled
|
||||
pub enable: bool,
|
||||
/// The broker URL
|
||||
pub broker: Url,
|
||||
/// The topic to publish to
|
||||
pub topic: String,
|
||||
/// The quality of service level
|
||||
pub qos: QoS,
|
||||
/// The username for the broker
|
||||
pub username: String,
|
||||
/// The password for the broker
|
||||
pub password: String,
|
||||
/// The maximum interval for reconnection attempts (Note: rumqttc has internal strategy)
|
||||
pub max_reconnect_interval: Duration,
|
||||
/// The keep alive interval
|
||||
pub keep_alive: Duration,
|
||||
/// The directory to store events in case of failure
|
||||
pub queue_dir: String,
|
||||
/// The maximum number of events to store
|
||||
pub queue_limit: u64,
|
||||
}
|
||||
|
||||
impl MQTTArgs {
|
||||
pub fn validate(&self) -> Result<(), TargetError> {
|
||||
if !self.enable {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
match self.broker.scheme() {
|
||||
"ws" | "wss" | "tcp" | "ssl" | "tls" | "tcps" | "mqtt" | "mqtts" => {}
|
||||
_ => {
|
||||
return Err(TargetError::Configuration("unknown protocol in broker address".to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
if !self.queue_dir.is_empty() {
|
||||
let path = std::path::Path::new(&self.queue_dir);
|
||||
if !path.is_absolute() {
|
||||
return Err(TargetError::Configuration("mqtt queueDir path should be absolute".to_string()));
|
||||
}
|
||||
|
||||
if self.qos == QoS::AtMostOnce {
|
||||
return Err(TargetError::Configuration(
|
||||
"QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queueDir is set".to_string(),
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
struct BgTaskManager {
|
||||
init_cell: OnceCell<tokio::task::JoinHandle<()>>,
|
||||
cancel_tx: mpsc::Sender<()>,
|
||||
initial_cancel_rx: Mutex<Option<mpsc::Receiver<()>>>,
|
||||
}
|
||||
|
||||
/// A target that sends events to an MQTT broker
|
||||
pub struct MQTTTarget {
|
||||
id: TargetID,
|
||||
args: MQTTArgs,
|
||||
client: Arc<Mutex<Option<AsyncClient>>>,
|
||||
store: Option<Box<dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync>>,
|
||||
connected: Arc<AtomicBool>,
|
||||
bg_task_manager: Arc<BgTaskManager>,
|
||||
}
|
||||
|
||||
impl MQTTTarget {
|
||||
/// Creates a new MQTTTarget
|
||||
#[instrument(skip(args), fields(target_id_as_string = %id))]
|
||||
pub fn new(id: String, args: MQTTArgs) -> Result<Self, TargetError> {
|
||||
args.validate()?;
|
||||
let target_id = TargetID::new(id.clone(), ChannelTargetType::Mqtt.as_str().to_string());
|
||||
let queue_store = if !args.queue_dir.is_empty() {
|
||||
let base_path = PathBuf::from(&args.queue_dir);
|
||||
let unique_dir_name = format!("rustfs-{}-{}", ChannelTargetType::Mqtt.as_str(), target_id.id).replace(":", "_");
|
||||
// Ensure the directory name is valid for filesystem
|
||||
let specific_queue_path = base_path.join(unique_dir_name);
|
||||
debug!(target_id = %target_id, path = %specific_queue_path.display(), "Initializing queue store for MQTT target");
|
||||
let store = crate::store::QueueStore::<Event>::new(specific_queue_path, args.queue_limit, STORE_EXTENSION);
|
||||
if let Err(e) = store.open() {
|
||||
error!(
|
||||
target_id = %target_id,
|
||||
error = %e,
|
||||
"Failed to open store for MQTT target"
|
||||
);
|
||||
return Err(TargetError::Storage(format!("{e}")));
|
||||
}
|
||||
Some(Box::new(store) as Box<dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync>)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let (cancel_tx, cancel_rx) = mpsc::channel(1);
|
||||
let bg_task_manager = Arc::new(BgTaskManager {
|
||||
init_cell: OnceCell::new(),
|
||||
cancel_tx,
|
||||
initial_cancel_rx: Mutex::new(Some(cancel_rx)),
|
||||
});
|
||||
|
||||
info!(target_id = %target_id, "MQTT target created");
|
||||
Ok(MQTTTarget {
|
||||
id: target_id,
|
||||
args,
|
||||
client: Arc::new(Mutex::new(None)),
|
||||
store: queue_store,
|
||||
connected: Arc::new(AtomicBool::new(false)),
|
||||
bg_task_manager,
|
||||
})
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(target_id = %self.id))]
|
||||
async fn init(&self) -> Result<(), TargetError> {
|
||||
if self.connected.load(Ordering::SeqCst) {
|
||||
debug!(target_id = %self.id, "Already connected.");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let bg_task_manager = Arc::clone(&self.bg_task_manager);
|
||||
let client_arc = Arc::clone(&self.client);
|
||||
let connected_arc = Arc::clone(&self.connected);
|
||||
let target_id_clone = self.id.clone();
|
||||
let args_clone = self.args.clone();
|
||||
|
||||
let _ = bg_task_manager
|
||||
.init_cell
|
||||
.get_or_try_init(|| async {
|
||||
debug!(target_id = %target_id_clone, "Initializing MQTT background task.");
|
||||
let host = args_clone.broker.host_str().unwrap_or("localhost");
|
||||
let port = args_clone.broker.port().unwrap_or(1883);
|
||||
let mut mqtt_options = MqttOptions::new(format!("rustfs_notify_{}", uuid::Uuid::new_v4()), host, port);
|
||||
mqtt_options
|
||||
.set_keep_alive(args_clone.keep_alive)
|
||||
.set_max_packet_size(100 * 1024 * 1024, 100 * 1024 * 1024); // 100MB
|
||||
|
||||
if !args_clone.username.is_empty() {
|
||||
mqtt_options.set_credentials(args_clone.username.clone(), args_clone.password.clone());
|
||||
}
|
||||
|
||||
let (new_client, eventloop) = AsyncClient::new(mqtt_options, 10);
|
||||
|
||||
if let Err(e) = new_client.subscribe(&args_clone.topic, args_clone.qos).await {
|
||||
error!(target_id = %target_id_clone, error = %e, "Failed to subscribe to MQTT topic during init");
|
||||
return Err(TargetError::Network(format!("MQTT subscribe failed: {e}")));
|
||||
}
|
||||
|
||||
let mut rx_guard = bg_task_manager.initial_cancel_rx.lock().await;
|
||||
let cancel_rx = rx_guard.take().ok_or_else(|| {
|
||||
error!(target_id = %target_id_clone, "MQTT cancel receiver already taken for task.");
|
||||
TargetError::Configuration("MQTT cancel receiver already taken for task".to_string())
|
||||
})?;
|
||||
drop(rx_guard);
|
||||
|
||||
*client_arc.lock().await = Some(new_client.clone());
|
||||
|
||||
info!(target_id = %target_id_clone, "Spawning MQTT event loop task.");
|
||||
let task_handle =
|
||||
tokio::spawn(run_mqtt_event_loop(eventloop, connected_arc.clone(), target_id_clone.clone(), cancel_rx));
|
||||
Ok(task_handle)
|
||||
})
|
||||
.await
|
||||
.map_err(|e: TargetError| {
|
||||
error!(target_id = %self.id, error = %e, "Failed to initialize MQTT background task");
|
||||
e
|
||||
})?;
|
||||
debug!(target_id = %self.id, "MQTT background task initialized successfully.");
|
||||
|
||||
match tokio::time::timeout(DEFAULT_CONNECTION_TIMEOUT, async {
|
||||
while !self.connected.load(Ordering::SeqCst) {
|
||||
if let Some(handle) = self.bg_task_manager.init_cell.get() {
|
||||
if handle.is_finished() && !self.connected.load(Ordering::SeqCst) {
|
||||
error!(target_id = %self.id, "MQTT background task exited prematurely before connection was established.");
|
||||
return Err(TargetError::Network("MQTT background task exited prematurely".to_string()));
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
debug!(target_id = %self.id, "MQTT target connected successfully.");
|
||||
Ok(())
|
||||
}).await {
|
||||
Ok(Ok(_)) => {
|
||||
info!(target_id = %self.id, "MQTT target initialized and connected.");
|
||||
Ok(())
|
||||
}
|
||||
Ok(Err(e)) => Err(e),
|
||||
Err(_) => {
|
||||
error!(target_id = %self.id, "Timeout waiting for MQTT connection after task spawn.");
|
||||
Err(TargetError::Network(
|
||||
"Timeout waiting for MQTT connection".to_string(),
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(skip(self, event), fields(target_id = %self.id))]
|
||||
async fn send(&self, event: &Event) -> Result<(), TargetError> {
|
||||
let client_guard = self.client.lock().await;
|
||||
let client = client_guard
|
||||
.as_ref()
|
||||
.ok_or_else(|| TargetError::Configuration("MQTT client not initialized".to_string()))?;
|
||||
|
||||
let object_name = urlencoding::decode(&event.s3.object.key)
|
||||
.map_err(|e| TargetError::Encoding(format!("Failed to decode object key: {e}")))?;
|
||||
|
||||
let key = format!("{}/{}", event.s3.bucket.name, object_name);
|
||||
|
||||
let log = EventLog {
|
||||
event_name: event.event_name,
|
||||
key,
|
||||
records: vec![event.clone()],
|
||||
};
|
||||
|
||||
let data = serde_json::to_vec(&log).map_err(|e| TargetError::Serialization(format!("Failed to serialize event: {e}")))?;
|
||||
|
||||
// Vec<u8> Convert to String, only for printing logs
|
||||
let data_string = String::from_utf8(data.clone())
|
||||
.map_err(|e| TargetError::Encoding(format!("Failed to convert event data to UTF-8: {e}")))?;
|
||||
debug!("Sending event to mqtt target: {}, event log: {}", self.id, data_string);
|
||||
|
||||
client
|
||||
.publish(&self.args.topic, self.args.qos, false, data)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
if e.to_string().contains("Connection") || e.to_string().contains("Timeout") {
|
||||
self.connected.store(false, Ordering::SeqCst);
|
||||
warn!(target_id = %self.id, error = %e, "Publish failed due to connection issue, marking as not connected.");
|
||||
TargetError::NotConnected
|
||||
} else {
|
||||
TargetError::Request(format!("Failed to publish message: {e}"))
|
||||
}
|
||||
})?;
|
||||
|
||||
debug!(target_id = %self.id, topic = %self.args.topic, "Event published to MQTT topic");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn clone_target(&self) -> Box<dyn Target + Send + Sync> {
|
||||
Box::new(MQTTTarget {
|
||||
id: self.id.clone(),
|
||||
args: self.args.clone(),
|
||||
client: self.client.clone(),
|
||||
store: self.store.as_ref().map(|s| s.boxed_clone()),
|
||||
connected: self.connected.clone(),
|
||||
bg_task_manager: self.bg_task_manager.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_mqtt_event_loop(
|
||||
mut eventloop: EventLoop,
|
||||
connected_status: Arc<AtomicBool>,
|
||||
target_id: TargetID,
|
||||
mut cancel_rx: mpsc::Receiver<()>,
|
||||
) {
|
||||
info!(target_id = %target_id, "MQTT event loop task started.");
|
||||
let mut initial_connection_established = false;
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
_ = cancel_rx.recv() => {
|
||||
info!(target_id = %target_id, "MQTT event loop task received cancellation signal. Shutting down.");
|
||||
break;
|
||||
}
|
||||
polled_event_result = async {
|
||||
if !initial_connection_established || !connected_status.load(Ordering::SeqCst) {
|
||||
match tokio::time::timeout(EVENT_LOOP_POLL_TIMEOUT, eventloop.poll()).await {
|
||||
Ok(Ok(event)) => Ok(event),
|
||||
Ok(Err(e)) => Err(e),
|
||||
Err(_) => {
|
||||
debug!(target_id = %target_id, "MQTT poll timed out (EVENT_LOOP_POLL_TIMEOUT) while not connected or status pending.");
|
||||
Err(rumqttc::ConnectionError::NetworkTimeout)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
eventloop.poll().await
|
||||
}
|
||||
} => {
|
||||
match polled_event_result {
|
||||
Ok(notification) => {
|
||||
trace!(target_id = %target_id, event = ?notification, "Received MQTT event");
|
||||
match notification {
|
||||
rumqttc::Event::Incoming(Packet::ConnAck(_conn_ack)) => {
|
||||
info!(target_id = %target_id, "MQTT connected (ConnAck).");
|
||||
connected_status.store(true, Ordering::SeqCst);
|
||||
initial_connection_established = true;
|
||||
}
|
||||
rumqttc::Event::Incoming(Packet::Publish(publish)) => {
|
||||
debug!(target_id = %target_id, topic = %publish.topic, payload_len = publish.payload.len(), "Received message on subscribed topic.");
|
||||
}
|
||||
rumqttc::Event::Incoming(Packet::Disconnect) => {
|
||||
info!(target_id = %target_id, "Received Disconnect packet from broker. MQTT connection lost.");
|
||||
connected_status.store(false, Ordering::SeqCst);
|
||||
}
|
||||
rumqttc::Event::Incoming(Packet::PingResp) => {
|
||||
trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive.");
|
||||
}
|
||||
rumqttc::Event::Incoming(Packet::SubAck(suback)) => {
|
||||
trace!(target_id = %target_id, "Received SubAck for pkid: {}", suback.pkid);
|
||||
}
|
||||
rumqttc::Event::Incoming(Packet::PubAck(puback)) => {
|
||||
trace!(target_id = %target_id, "Received PubAck for pkid: {}", puback.pkid);
|
||||
}
|
||||
// Process other incoming packet types as needed (PubRec, PubRel, PubComp, UnsubAck)
|
||||
rumqttc::Event::Outgoing(Outgoing::Disconnect) => {
|
||||
info!(target_id = %target_id, "MQTT outgoing disconnect initiated by client.");
|
||||
connected_status.store(false, Ordering::SeqCst);
|
||||
}
|
||||
rumqttc::Event::Outgoing(Outgoing::PingReq) => {
|
||||
trace!(target_id = %target_id, "Client sent PingReq to broker.");
|
||||
}
|
||||
// Other Outgoing events (Subscribe, Unsubscribe, Publish) usually do not need to handle connection status here,
|
||||
// Because they are actions initiated by the client.
|
||||
_ => {
|
||||
// Log other unspecified MQTT events that are not handled, which helps debug
|
||||
trace!(target_id = %target_id, "Unhandled or generic MQTT event: {:?}", notification);
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
connected_status.store(false, Ordering::SeqCst);
|
||||
error!(target_id = %target_id, error = %e, "Error from MQTT event loop poll");
|
||||
|
||||
if matches!(e, rumqttc::ConnectionError::NetworkTimeout) && (!initial_connection_established || !connected_status.load(Ordering::SeqCst)) {
|
||||
warn!(target_id = %target_id, "Timeout during initial poll or pending state, will retry.");
|
||||
continue;
|
||||
}
|
||||
|
||||
if matches!(e,
|
||||
ConnectionError::Io(_) |
|
||||
ConnectionError::NetworkTimeout |
|
||||
ConnectionError::ConnectionRefused(_) |
|
||||
ConnectionError::Tls(_)
|
||||
) {
|
||||
warn!(target_id = %target_id, error = %e, "MQTT connection error. Relying on rumqttc for reconnection if applicable.");
|
||||
}
|
||||
// Here you can decide whether to break loops based on the error type.
|
||||
// For example, for some unrecoverable errors.
|
||||
if is_fatal_mqtt_error(&e) {
|
||||
error!(target_id = %target_id, error = %e, "Fatal MQTT error, terminating event loop.");
|
||||
break;
|
||||
}
|
||||
// rumqttc's eventloop.poll() may return Err and terminate after some errors,
|
||||
// Or it will handle reconnection internally. The continue here will make select! wait again.
|
||||
// If the error is temporary and rumqttc is handling reconnection, poll() should eventually succeed or return a different error again.
|
||||
// Sleep briefly to avoid busy cycles in case of rapid failure.
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
connected_status.store(false, Ordering::SeqCst);
|
||||
info!(target_id = %target_id, "MQTT event loop task finished.");
|
||||
}
|
||||
|
||||
/// Check whether the given MQTT connection error should be considered a fatal error,
|
||||
/// For fatal errors, the event loop should terminate.
|
||||
fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
|
||||
match err {
|
||||
// If the client request has been processed all (for example, AsyncClient is dropped), the event loop can end.
|
||||
ConnectionError::RequestsDone => true,
|
||||
|
||||
// Check for the underlying MQTT status error
|
||||
ConnectionError::MqttState(state_err) => {
|
||||
// The type of state_err is &rumqttc::StateError
|
||||
match state_err {
|
||||
// If StateError is caused by deserialization issues, check the underlying MqttBytesError
|
||||
rumqttc::StateError::Deserialization(mqtt_bytes_err) => { // The type of mqtt_bytes_err is &rumqttc::mqttbytes::Error
|
||||
matches!(
|
||||
mqtt_bytes_err,
|
||||
MqttBytesError::InvalidProtocol // Invalid agreement
|
||||
| MqttBytesError::InvalidProtocolLevel(_) // Invalid protocol level
|
||||
| MqttBytesError::IncorrectPacketFormat // Package format is incorrect
|
||||
| MqttBytesError::InvalidPacketType(_) // Invalid package type
|
||||
| MqttBytesError::MalformedPacket // Package format error
|
||||
| MqttBytesError::PayloadTooLong // Too long load
|
||||
| MqttBytesError::PayloadSizeLimitExceeded(_) // Load size limit exceeded
|
||||
| MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation)
|
||||
)
|
||||
}
|
||||
// Others that are fatal StateError variants
|
||||
rumqttc::StateError::InvalidState // The internal state machine is in invalid state
|
||||
| rumqttc::StateError::WrongPacket // Agreement Violation: Unexpected Data Packet Received
|
||||
| rumqttc::StateError::Unsolicited(_) // Agreement Violation: Unsolicited ACK Received
|
||||
| rumqttc::StateError::OutgoingPacketTooLarge { .. } // Try to send too large packets
|
||||
| rumqttc::StateError::EmptySubscription // Agreement violation (if this stage occurs)
|
||||
=> true,
|
||||
|
||||
// Other StateErrors (such as Io, AwaitPingResp, CollisionTimeout) are not considered deadly here.
|
||||
// They may be processed internally by rumqttc or upgraded to other ConnectionError types.
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
// Other types of ConnectionErrors (such as Io, Tls, NetworkTimeout, ConnectionRefused, NotConnAck, etc.)
|
||||
// It is usually considered temporary, or the reconnect logic inside rumqttc will be processed.
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Target for MQTTTarget {
|
||||
fn id(&self) -> TargetID {
|
||||
self.id.clone()
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(target_id = %self.id))]
|
||||
async fn is_active(&self) -> Result<bool, TargetError> {
|
||||
debug!(target_id = %self.id, "Checking if MQTT target is active.");
|
||||
if self.client.lock().await.is_none() && !self.connected.load(Ordering::SeqCst) {
|
||||
// Check if the background task is running and has not panicked
|
||||
if let Some(handle) = self.bg_task_manager.init_cell.get() {
|
||||
if handle.is_finished() {
|
||||
error!(target_id = %self.id, "MQTT background task has finished, possibly due to an error. Target is not active.");
|
||||
return Err(TargetError::Network("MQTT background task terminated".to_string()));
|
||||
}
|
||||
}
|
||||
debug!(target_id = %self.id, "MQTT client not yet initialized or task not running/connected.");
|
||||
return Err(TargetError::Configuration(
|
||||
"MQTT client not available or not initialized/connected".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
if self.connected.load(Ordering::SeqCst) {
|
||||
debug!(target_id = %self.id, "MQTT target is active (connected flag is true).");
|
||||
Ok(true)
|
||||
} else {
|
||||
debug!(target_id = %self.id, "MQTT target is not connected (connected flag is false).");
|
||||
Err(TargetError::NotConnected)
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(skip(self, event), fields(target_id = %self.id))]
|
||||
async fn save(&self, event: Arc<Event>) -> Result<(), TargetError> {
|
||||
if let Some(store) = &self.store {
|
||||
debug!(target_id = %self.id, "Event saved to store start");
|
||||
// If store is configured, ONLY put the event into the store.
|
||||
// Do NOT send it directly here.
|
||||
match store.put(event.clone()) {
|
||||
Ok(_) => {
|
||||
debug!(target_id = %self.id, "Event saved to store for MQTT target successfully.");
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
error!(target_id = %self.id, error = %e, "Failed to save event to store");
|
||||
return Err(TargetError::Storage(format!("Failed to save event to store: {e}")));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if !self.is_enabled() {
|
||||
return Err(TargetError::Disabled);
|
||||
}
|
||||
|
||||
if !self.connected.load(Ordering::SeqCst) {
|
||||
warn!(target_id = %self.id, "Attempting to send directly but not connected; trying to init.");
|
||||
// Call the struct's init method, not the trait's default
|
||||
match MQTTTarget::init(self).await {
|
||||
Ok(_) => debug!(target_id = %self.id, "MQTT target initialized successfully."),
|
||||
Err(e) => {
|
||||
error!(target_id = %self.id, error = %e, "Failed to initialize MQTT target.");
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
}
|
||||
if !self.connected.load(Ordering::SeqCst) {
|
||||
error!(target_id = %self.id, "Cannot save (send directly) as target is not active after init attempt.");
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
}
|
||||
self.send(&event).await
|
||||
}
|
||||
}
|
||||
|
||||
#[instrument(skip(self), fields(target_id = %self.id))]
|
||||
async fn send_from_store(&self, key: Key) -> Result<(), TargetError> {
|
||||
debug!(target_id = %self.id, ?key, "Attempting to send event from store with key.");
|
||||
|
||||
if !self.is_enabled() {
|
||||
return Err(TargetError::Disabled);
|
||||
}
|
||||
|
||||
if !self.connected.load(Ordering::SeqCst) {
|
||||
warn!(target_id = %self.id, "Not connected; trying to init before sending from store.");
|
||||
match MQTTTarget::init(self).await {
|
||||
Ok(_) => debug!(target_id = %self.id, "MQTT target initialized successfully."),
|
||||
Err(e) => {
|
||||
error!(target_id = %self.id, error = %e, "Failed to initialize MQTT target.");
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
}
|
||||
if !self.connected.load(Ordering::SeqCst) {
|
||||
error!(target_id = %self.id, "Cannot send from store as target is not active after init attempt.");
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
}
|
||||
|
||||
let store = self
|
||||
.store
|
||||
.as_ref()
|
||||
.ok_or_else(|| TargetError::Configuration("No store configured".to_string()))?;
|
||||
|
||||
let event = match store.get(&key) {
|
||||
Ok(event) => {
|
||||
debug!(target_id = %self.id, ?key, "Retrieved event from store for sending.");
|
||||
event
|
||||
}
|
||||
Err(StoreError::NotFound) => {
|
||||
// Assuming NotFound takes the key
|
||||
debug!(target_id = %self.id, ?key, "Event not found in store for sending.");
|
||||
return Ok(());
|
||||
}
|
||||
Err(e) => {
|
||||
error!(
|
||||
target_id = %self.id,
|
||||
error = %e,
|
||||
"Failed to get event from store"
|
||||
);
|
||||
return Err(TargetError::Storage(format!("Failed to get event from store: {e}")));
|
||||
}
|
||||
};
|
||||
|
||||
debug!(target_id = %self.id, ?key, "Sending event from store.");
|
||||
if let Err(e) = self.send(&event).await {
|
||||
if matches!(e, TargetError::NotConnected) {
|
||||
warn!(target_id = %self.id, "Failed to send event from store: Not connected. Event remains in store.");
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
error!(target_id = %self.id, error = %e, "Failed to send event from store with an unexpected error.");
|
||||
return Err(e);
|
||||
}
|
||||
debug!(target_id = %self.id, ?key, "Event sent from store successfully. deleting from store. ");
|
||||
|
||||
match store.del(&key) {
|
||||
Ok(_) => {
|
||||
debug!(target_id = %self.id, ?key, "Event deleted from store after successful send.")
|
||||
}
|
||||
Err(StoreError::NotFound) => {
|
||||
debug!(target_id = %self.id, ?key, "Event already deleted from store.");
|
||||
}
|
||||
Err(e) => {
|
||||
error!(target_id = %self.id, error = %e, "Failed to delete event from store after send.");
|
||||
return Err(TargetError::Storage(format!("Failed to delete event from store: {e}")));
|
||||
}
|
||||
}
|
||||
|
||||
debug!(target_id = %self.id, ?key, "Event deleted from store.");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn close(&self) -> Result<(), TargetError> {
|
||||
info!(target_id = %self.id, "Attempting to close MQTT target.");
|
||||
|
||||
if let Err(e) = self.bg_task_manager.cancel_tx.send(()).await {
|
||||
warn!(target_id = %self.id, error = %e, "Failed to send cancel signal to MQTT background task. It might have already exited.");
|
||||
}
|
||||
|
||||
// Wait for the task to finish if it was initialized
|
||||
if let Some(_task_handle) = self.bg_task_manager.init_cell.get() {
|
||||
debug!(target_id = %self.id, "Waiting for MQTT background task to complete...");
|
||||
// It's tricky to await here if close is called from a sync context or Drop
|
||||
// For async close, this is fine. Consider a timeout.
|
||||
// let _ = tokio::time::timeout(Duration::from_secs(5), task_handle.await).await;
|
||||
// If task_handle.await is directly used, ensure it's not awaited multiple times if close can be called multiple times.
|
||||
// For now, we rely on the signal and the task's self-termination.
|
||||
}
|
||||
|
||||
if let Some(client_instance) = self.client.lock().await.take() {
|
||||
info!(target_id = %self.id, "Disconnecting MQTT client.");
|
||||
if let Err(e) = client_instance.disconnect().await {
|
||||
warn!(target_id = %self.id, error = %e, "Error during MQTT client disconnect.");
|
||||
}
|
||||
}
|
||||
|
||||
self.connected.store(false, Ordering::SeqCst);
|
||||
info!(target_id = %self.id, "MQTT target close method finished.");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn store(&self) -> Option<&(dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync)> {
|
||||
self.store.as_deref()
|
||||
}
|
||||
|
||||
fn clone_dyn(&self) -> Box<dyn Target + Send + Sync> {
|
||||
self.clone_target()
|
||||
}
|
||||
|
||||
async fn init(&self) -> Result<(), TargetError> {
|
||||
if !self.is_enabled() {
|
||||
debug!(target_id = %self.id, "Target is disabled, skipping init.");
|
||||
return Ok(());
|
||||
}
|
||||
// Call the internal init logic
|
||||
MQTTTarget::init(self).await
|
||||
}
|
||||
|
||||
fn is_enabled(&self) -> bool {
|
||||
self.args.enable
|
||||
}
|
||||
}
|
||||
@@ -1,407 +0,0 @@
|
||||
// Copyright 2024 RustFS Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::target::ChannelTargetType;
|
||||
use crate::{
|
||||
StoreError, Target,
|
||||
arn::TargetID,
|
||||
error::TargetError,
|
||||
event::{Event, EventLog},
|
||||
store::{Key, Store},
|
||||
};
|
||||
use async_trait::async_trait;
|
||||
use reqwest::{Client, StatusCode, Url};
|
||||
use rustfs_config::notify::STORE_EXTENSION;
|
||||
use std::{
|
||||
path::PathBuf,
|
||||
sync::{
|
||||
Arc,
|
||||
atomic::{AtomicBool, Ordering},
|
||||
},
|
||||
time::Duration,
|
||||
};
|
||||
use tokio::net::lookup_host;
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{debug, error, info, instrument};
|
||||
use urlencoding;
|
||||
|
||||
/// Arguments for configuring a Webhook target
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct WebhookArgs {
|
||||
/// Whether the target is enabled
|
||||
pub enable: bool,
|
||||
/// The endpoint URL to send events to
|
||||
pub endpoint: Url,
|
||||
/// The authorization token for the endpoint
|
||||
pub auth_token: String,
|
||||
/// The directory to store events in case of failure
|
||||
pub queue_dir: String,
|
||||
/// The maximum number of events to store
|
||||
pub queue_limit: u64,
|
||||
/// The client certificate for TLS (PEM format)
|
||||
pub client_cert: String,
|
||||
/// The client key for TLS (PEM format)
|
||||
pub client_key: String,
|
||||
}
|
||||
|
||||
impl WebhookArgs {
|
||||
/// WebhookArgs verification method
|
||||
pub fn validate(&self) -> Result<(), TargetError> {
|
||||
if !self.enable {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if self.endpoint.as_str().is_empty() {
|
||||
return Err(TargetError::Configuration("endpoint empty".to_string()));
|
||||
}
|
||||
|
||||
if !self.queue_dir.is_empty() {
|
||||
let path = std::path::Path::new(&self.queue_dir);
|
||||
if !path.is_absolute() {
|
||||
return Err(TargetError::Configuration("webhook queueDir path should be absolute".to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
if !self.client_cert.is_empty() && self.client_key.is_empty()
|
||||
|| self.client_cert.is_empty() && !self.client_key.is_empty()
|
||||
{
|
||||
return Err(TargetError::Configuration("cert and key must be specified as a pair".to_string()));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// A target that sends events to a webhook
|
||||
pub struct WebhookTarget {
|
||||
id: TargetID,
|
||||
args: WebhookArgs,
|
||||
http_client: Arc<Client>,
|
||||
// Add Send + Sync constraints to ensure thread safety
|
||||
store: Option<Box<dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync>>,
|
||||
initialized: AtomicBool,
|
||||
addr: String,
|
||||
cancel_sender: mpsc::Sender<()>,
|
||||
}
|
||||
|
||||
impl WebhookTarget {
|
||||
/// Clones the WebhookTarget, creating a new instance with the same configuration
|
||||
pub fn clone_box(&self) -> Box<dyn Target + Send + Sync> {
|
||||
Box::new(WebhookTarget {
|
||||
id: self.id.clone(),
|
||||
args: self.args.clone(),
|
||||
http_client: Arc::clone(&self.http_client),
|
||||
store: self.store.as_ref().map(|s| s.boxed_clone()),
|
||||
initialized: AtomicBool::new(self.initialized.load(Ordering::SeqCst)),
|
||||
addr: self.addr.clone(),
|
||||
cancel_sender: self.cancel_sender.clone(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Creates a new WebhookTarget
|
||||
#[instrument(skip(args), fields(target_id = %id))]
|
||||
pub fn new(id: String, args: WebhookArgs) -> Result<Self, TargetError> {
|
||||
// First verify the parameters
|
||||
args.validate()?;
|
||||
// Create a TargetID
|
||||
let target_id = TargetID::new(id, ChannelTargetType::Webhook.as_str().to_string());
|
||||
// Build HTTP client
|
||||
let mut client_builder = Client::builder()
|
||||
.timeout(Duration::from_secs(30))
|
||||
.user_agent(rustfs_utils::sys::get_user_agent(rustfs_utils::sys::ServiceType::Basis));
|
||||
|
||||
// Supplementary certificate processing logic
|
||||
if !args.client_cert.is_empty() && !args.client_key.is_empty() {
|
||||
// Add client certificate
|
||||
let cert = std::fs::read(&args.client_cert)
|
||||
.map_err(|e| TargetError::Configuration(format!("Failed to read client cert: {e}")))?;
|
||||
let key = std::fs::read(&args.client_key)
|
||||
.map_err(|e| TargetError::Configuration(format!("Failed to read client key: {e}")))?;
|
||||
|
||||
let identity = reqwest::Identity::from_pem(&[cert, key].concat())
|
||||
.map_err(|e| TargetError::Configuration(format!("Failed to create identity: {e}")))?;
|
||||
client_builder = client_builder.identity(identity);
|
||||
}
|
||||
|
||||
let http_client = Arc::new(
|
||||
client_builder
|
||||
.build()
|
||||
.map_err(|e| TargetError::Configuration(format!("Failed to build HTTP client: {e}")))?,
|
||||
);
|
||||
|
||||
// Build storage
|
||||
let queue_store = if !args.queue_dir.is_empty() {
|
||||
let queue_dir =
|
||||
PathBuf::from(&args.queue_dir).join(format!("rustfs-{}-{}", ChannelTargetType::Webhook.as_str(), target_id.id));
|
||||
let store = crate::store::QueueStore::<Event>::new(queue_dir, args.queue_limit, STORE_EXTENSION);
|
||||
|
||||
if let Err(e) = store.open() {
|
||||
error!("Failed to open store for Webhook target {}: {}", target_id.id, e);
|
||||
return Err(TargetError::Storage(format!("{e}")));
|
||||
}
|
||||
|
||||
// Make sure that the Store trait implemented by QueueStore matches the expected error type
|
||||
Some(Box::new(store) as Box<dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync>)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// resolved address
|
||||
let addr = {
|
||||
let host = args.endpoint.host_str().unwrap_or("localhost");
|
||||
let port = args
|
||||
.endpoint
|
||||
.port()
|
||||
.unwrap_or_else(|| if args.endpoint.scheme() == "https" { 443 } else { 80 });
|
||||
format!("{host}:{port}")
|
||||
};
|
||||
|
||||
// Create a cancel channel
|
||||
let (cancel_sender, _) = mpsc::channel(1);
|
||||
info!(target_id = %target_id.id, "Webhook target created");
|
||||
Ok(WebhookTarget {
|
||||
id: target_id,
|
||||
args,
|
||||
http_client,
|
||||
store: queue_store,
|
||||
initialized: AtomicBool::new(false),
|
||||
addr,
|
||||
cancel_sender,
|
||||
})
|
||||
}
|
||||
|
||||
async fn init(&self) -> Result<(), TargetError> {
|
||||
// Use CAS operations to ensure thread-safe initialization
|
||||
if !self.initialized.load(Ordering::SeqCst) {
|
||||
// Check the connection
|
||||
match self.is_active().await {
|
||||
Ok(true) => {
|
||||
info!("Webhook target {} is active", self.id);
|
||||
}
|
||||
Ok(false) => {
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to check if Webhook target {} is active: {}", self.id, e);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
self.initialized.store(true, Ordering::SeqCst);
|
||||
info!("Webhook target {} initialized", self.id);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn send(&self, event: &Event) -> Result<(), TargetError> {
|
||||
info!("Webhook Sending event to webhook target: {}", self.id);
|
||||
let object_name = urlencoding::decode(&event.s3.object.key)
|
||||
.map_err(|e| TargetError::Encoding(format!("Failed to decode object key: {e}")))?;
|
||||
|
||||
let key = format!("{}/{}", event.s3.bucket.name, object_name);
|
||||
|
||||
let log = EventLog {
|
||||
event_name: event.event_name,
|
||||
key,
|
||||
records: vec![event.clone()],
|
||||
};
|
||||
|
||||
let data = serde_json::to_vec(&log).map_err(|e| TargetError::Serialization(format!("Failed to serialize event: {e}")))?;
|
||||
|
||||
// Vec<u8> Convert to String
|
||||
let data_string = String::from_utf8(data.clone())
|
||||
.map_err(|e| TargetError::Encoding(format!("Failed to convert event data to UTF-8: {e}")))?;
|
||||
debug!("Sending event to webhook target: {}, event log: {}", self.id, data_string);
|
||||
|
||||
// build request
|
||||
let mut req_builder = self
|
||||
.http_client
|
||||
.post(self.args.endpoint.as_str())
|
||||
.header("Content-Type", "application/json");
|
||||
|
||||
if !self.args.auth_token.is_empty() {
|
||||
// Split auth_token string to check if the authentication type is included
|
||||
let tokens: Vec<&str> = self.args.auth_token.split_whitespace().collect();
|
||||
match tokens.len() {
|
||||
2 => {
|
||||
// Already include authentication type and token, such as "Bearer token123"
|
||||
req_builder = req_builder.header("Authorization", &self.args.auth_token);
|
||||
}
|
||||
1 => {
|
||||
// Only tokens, need to add "Bearer" prefix
|
||||
req_builder = req_builder.header("Authorization", format!("Bearer {}", self.args.auth_token));
|
||||
}
|
||||
_ => {
|
||||
// Empty string or other situations, no authentication header is added
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send a request
|
||||
let resp = req_builder.body(data).send().await.map_err(|e| {
|
||||
if e.is_timeout() || e.is_connect() {
|
||||
TargetError::NotConnected
|
||||
} else {
|
||||
TargetError::Request(format!("Failed to send request: {e}"))
|
||||
}
|
||||
})?;
|
||||
|
||||
let status = resp.status();
|
||||
if status.is_success() {
|
||||
debug!("Event sent to webhook target: {}", self.id);
|
||||
Ok(())
|
||||
} else if status == StatusCode::FORBIDDEN {
|
||||
Err(TargetError::Authentication(format!(
|
||||
"{} returned '{}', please check if your auth token is correctly set",
|
||||
self.args.endpoint, status
|
||||
)))
|
||||
} else {
|
||||
Err(TargetError::Request(format!(
|
||||
"{} returned '{}', please check your endpoint configuration",
|
||||
self.args.endpoint, status
|
||||
)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Target for WebhookTarget {
|
||||
fn id(&self) -> TargetID {
|
||||
self.id.clone()
|
||||
}
|
||||
|
||||
// Make sure Future is Send
|
||||
async fn is_active(&self) -> Result<bool, TargetError> {
|
||||
let socket_addr = lookup_host(&self.addr)
|
||||
.await
|
||||
.map_err(|e| TargetError::Network(format!("Failed to resolve host: {e}")))?
|
||||
.next()
|
||||
.ok_or_else(|| TargetError::Network("No address found".to_string()))?;
|
||||
debug!("is_active socket addr: {},target id:{}", socket_addr, self.id.id);
|
||||
match tokio::time::timeout(Duration::from_secs(5), tokio::net::TcpStream::connect(socket_addr)).await {
|
||||
Ok(Ok(_)) => {
|
||||
debug!("Connection to {} is active", self.addr);
|
||||
Ok(true)
|
||||
}
|
||||
Ok(Err(e)) => {
|
||||
debug!("Connection to {} failed: {}", self.addr, e);
|
||||
if e.kind() == std::io::ErrorKind::ConnectionRefused {
|
||||
Err(TargetError::NotConnected)
|
||||
} else {
|
||||
Err(TargetError::Network(format!("Connection failed: {e}")))
|
||||
}
|
||||
}
|
||||
Err(_) => Err(TargetError::Timeout("Connection timed out".to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
async fn save(&self, event: Arc<Event>) -> Result<(), TargetError> {
|
||||
if let Some(store) = &self.store {
|
||||
// Call the store method directly, no longer need to acquire the lock
|
||||
store
|
||||
.put(event)
|
||||
.map_err(|e| TargetError::Storage(format!("Failed to save event to store: {e}")))?;
|
||||
debug!("Event saved to store for target: {}", self.id);
|
||||
Ok(())
|
||||
} else {
|
||||
match self.init().await {
|
||||
Ok(_) => (),
|
||||
Err(e) => {
|
||||
error!("Failed to initialize Webhook target {}: {}", self.id.id, e);
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
}
|
||||
self.send(&event).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn send_from_store(&self, key: Key) -> Result<(), TargetError> {
|
||||
debug!("Sending event from store for target: {}", self.id);
|
||||
match self.init().await {
|
||||
Ok(_) => {
|
||||
debug!("Event sent to store for target: {}", self.name());
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to initialize Webhook target {}: {}", self.id.id, e);
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
}
|
||||
|
||||
let store = self
|
||||
.store
|
||||
.as_ref()
|
||||
.ok_or_else(|| TargetError::Configuration("No store configured".to_string()))?;
|
||||
|
||||
// Get events directly from the store, no longer need to acquire locks
|
||||
let event = match store.get(&key) {
|
||||
Ok(event) => event,
|
||||
Err(StoreError::NotFound) => return Ok(()),
|
||||
Err(e) => {
|
||||
return Err(TargetError::Storage(format!("Failed to get event from store: {e}")));
|
||||
}
|
||||
};
|
||||
|
||||
if let Err(e) = self.send(&event).await {
|
||||
if let TargetError::NotConnected = e {
|
||||
return Err(TargetError::NotConnected);
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
// Use the immutable reference of the store to delete the event content corresponding to the key
|
||||
debug!("Deleting event from store for target: {}, key:{}, start", self.id, key.to_string());
|
||||
match store.del(&key) {
|
||||
Ok(_) => debug!("Event deleted from store for target: {}, key:{}, end", self.id, key.to_string()),
|
||||
Err(e) => {
|
||||
error!("Failed to delete event from store: {}", e);
|
||||
return Err(TargetError::Storage(format!("Failed to delete event from store: {e}")));
|
||||
}
|
||||
}
|
||||
|
||||
debug!("Event sent from store and deleted for target: {}", self.id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn close(&self) -> Result<(), TargetError> {
|
||||
// Send cancel signal to background tasks
|
||||
let _ = self.cancel_sender.try_send(());
|
||||
info!("Webhook target closed: {}", self.id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn store(&self) -> Option<&(dyn Store<Event, Error = StoreError, Key = Key> + Send + Sync)> {
|
||||
// Returns the reference to the internal store
|
||||
self.store.as_deref()
|
||||
}
|
||||
|
||||
fn clone_dyn(&self) -> Box<dyn Target + Send + Sync> {
|
||||
self.clone_box()
|
||||
}
|
||||
|
||||
// The existing init method can meet the needs well, but we need to make sure it complies with the Target trait
|
||||
// We can use the existing init method, but adjust the return value to match the trait requirement
|
||||
async fn init(&self) -> Result<(), TargetError> {
|
||||
// If the target is disabled, return to success directly
|
||||
if !self.is_enabled() {
|
||||
debug!("Webhook target {} is disabled, skipping initialization", self.id);
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
// Use existing initialization logic
|
||||
WebhookTarget::init(self).await
|
||||
}
|
||||
|
||||
fn is_enabled(&self) -> bool {
|
||||
self.args.enable
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user