feat(targets): add NATS and Pulsar target support (#2618)

crates/targets/src/check.rs

@@ -92,3 +92,39 @@ pub async fn check_mqtt_broker_available_with_tls(
         Err(_) => Err(crate::TargetError::Timeout("MQTT connection timed out".to_string())),
     }
 }
+
+pub async fn check_nats_server_available(args: &crate::target::nats::NATSArgs) -> Result<(), crate::TargetError> {
+    match tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        let client = crate::target::nats::connect_nats(args).await?;
+        client
+            .flush()
+            .await
+            .map_err(|e| crate::TargetError::Network(format!("NATS connection check failed: {e}")))?;
+        client
+            .drain()
+            .await
+            .map_err(|e| crate::TargetError::Network(format!("Failed to close NATS check connection: {e}")))?;
+        Ok(())
+    })
+    .await
+    {
+        Ok(result) => result,
+        Err(_) => Err(crate::TargetError::Timeout("NATS connection timed out".to_string())),
+    }
+}
+
+pub async fn check_pulsar_broker_available(args: &crate::target::pulsar::PulsarArgs) -> Result<(), crate::TargetError> {
+    match tokio::time::timeout(std::time::Duration::from_secs(5), async {
+        let client = crate::target::pulsar::connect_pulsar(args).await?;
+        client
+            .lookup_partitioned_topic(args.topic.clone())
+            .await
+            .map_err(|e| crate::TargetError::Network(format!("Pulsar topic lookup failed: {e}")))?;
+        Ok(())
+    })
+    .await
+    {
+        Ok(result) => result,
+        Err(_) => Err(crate::TargetError::Timeout("Pulsar connection timed out".to_string())),
+    }
+}

crates/targets/src/config/common.rs

@@ -0,0 +1,182 @@
+// 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 crate::target::pulsar::validate_pulsar_broker;
+use async_nats::ServerAddr;
+use rustfs_config::{
+    DEFAULT_DELIMITER, ENABLE_KEY, EnableState, NATS_CREDENTIALS_FILE, NATS_PASSWORD, NATS_QUEUE_DIR, NATS_SUBJECT, NATS_TLS_CA,
+    NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY, NATS_TOKEN, NATS_USERNAME, PULSAR_AUTH_TOKEN, PULSAR_PASSWORD, PULSAR_QUEUE_DIR,
+    PULSAR_TLS_ALLOW_INSECURE, PULSAR_TLS_CA, PULSAR_TLS_HOSTNAME_VERIFICATION, PULSAR_TOPIC, PULSAR_USERNAME,
+};
+use rustfs_ecstore::config::KVS;
+use std::collections::HashSet;
+use std::path::Path;
+use std::str::FromStr;
+use url::Url;
+
+pub(super) fn split_env_field_and_instance(rest: &str, valid_fields: &HashSet<String>) -> Option<(String, String)> {
+    let normalized = rest.to_lowercase();
+    if valid_fields.contains(&normalized) {
+        return Some((normalized, DEFAULT_DELIMITER.to_string()));
+    }
+
+    valid_fields
+        .iter()
+        .filter_map(|field| {
+            normalized
+                .strip_prefix(field)
+                .and_then(|suffix| suffix.strip_prefix(DEFAULT_DELIMITER))
+                .filter(|instance_id| !instance_id.is_empty())
+                .map(|instance_id| (field.clone(), instance_id.to_string()))
+        })
+        .max_by_key(|(field, _)| field.len())
+}
+
+pub(super) fn is_target_enabled(config: &KVS) -> bool {
+    config
+        .lookup(ENABLE_KEY)
+        .map(|v| {
+            EnableState::from_str(v.as_str())
+                .ok()
+                .map(|s| s.is_enabled())
+                .unwrap_or(false)
+        })
+        .unwrap_or(false)
+}
+
+pub(super) fn parse_target_bool(value: Option<&str>) -> Option<bool> {
+    let value = value?.trim();
+    if value.is_empty() {
+        return None;
+    }
+    value
+        .parse::<EnableState>()
+        .map(EnableState::is_enabled)
+        .or_else(|_| value.parse::<bool>())
+        .ok()
+}
+
+pub(super) fn validate_nats_server_config(server: &ServerAddr, config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
+    if config.lookup(NATS_SUBJECT).unwrap_or_default().trim().is_empty() {
+        return Err(TargetError::Configuration("Missing NATS subject".to_string()));
+    }
+
+    if server.has_user_pass() {
+        return Err(TargetError::Configuration("NATS address must not embed username or password".to_string()));
+    }
+
+    let username = config.lookup(NATS_USERNAME).unwrap_or_default();
+    let password = config.lookup(NATS_PASSWORD).unwrap_or_default();
+    let token = config.lookup(NATS_TOKEN).unwrap_or_default();
+    let credentials_file = config.lookup(NATS_CREDENTIALS_FILE).unwrap_or_default();
+
+    let mut auth_methods = 0usize;
+    if !token.is_empty() {
+        auth_methods += 1;
+    }
+    if !credentials_file.is_empty() {
+        auth_methods += 1;
+        if !Path::new(&credentials_file).is_absolute() {
+            return Err(TargetError::Configuration(format!("{NATS_CREDENTIALS_FILE} must be an absolute path")));
+        }
+    }
+    if !username.is_empty() || !password.is_empty() {
+        if username.is_empty() != password.is_empty() {
+            return Err(TargetError::Configuration(
+                "NATS username and password must be specified together".to_string(),
+            ));
+        }
+        auth_methods += 1;
+    }
+    if auth_methods > 1 {
+        return Err(TargetError::Configuration(
+            "NATS supports only one auth method at a time: token, username/password, or credentials_file".to_string(),
+        ));
+    }
+
+    let tls_ca = config.lookup(NATS_TLS_CA).unwrap_or_default();
+    let tls_client_cert = config.lookup(NATS_TLS_CLIENT_CERT).unwrap_or_default();
+    let tls_client_key = config.lookup(NATS_TLS_CLIENT_KEY).unwrap_or_default();
+    if !tls_ca.is_empty() && !Path::new(&tls_ca).is_absolute() {
+        return Err(TargetError::Configuration(format!("{NATS_TLS_CA} must be an absolute path")));
+    }
+    if !tls_client_cert.is_empty() && !Path::new(&tls_client_cert).is_absolute() {
+        return Err(TargetError::Configuration(format!("{NATS_TLS_CLIENT_CERT} must be an absolute path")));
+    }
+    if !tls_client_key.is_empty() && !Path::new(&tls_client_key).is_absolute() {
+        return Err(TargetError::Configuration(format!("{NATS_TLS_CLIENT_KEY} must be an absolute path")));
+    }
+    if tls_client_cert.is_empty() != tls_client_key.is_empty() {
+        return Err(TargetError::Configuration(
+            "NATS tls_client_cert and tls_client_key must be specified together".to_string(),
+        ));
+    }
+
+    let queue_dir = config.lookup(NATS_QUEUE_DIR).unwrap_or_else(|| default_queue_dir.to_string());
+    if !queue_dir.is_empty() && !Path::new(&queue_dir).is_absolute() {
+        return Err(TargetError::Configuration("NATS queue directory must be an absolute path".to_string()));
+    }
+
+    let _ = server;
+    Ok(())
+}
+
+pub(super) fn validate_pulsar_broker_config(broker: &str, config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
+    let url = validate_pulsar_broker(broker)?;
+
+    if config.lookup(PULSAR_TOPIC).unwrap_or_default().trim().is_empty() {
+        return Err(TargetError::Configuration("Missing Pulsar topic".to_string()));
+    }
+
+    let auth_token = config.lookup(PULSAR_AUTH_TOKEN).unwrap_or_default();
+    let username = config.lookup(PULSAR_USERNAME).unwrap_or_default();
+    let password = config.lookup(PULSAR_PASSWORD).unwrap_or_default();
+    if !auth_token.is_empty() && (!username.is_empty() || !password.is_empty()) {
+        return Err(TargetError::Configuration(
+            "Pulsar supports either auth_token or username/password auth, not both".to_string(),
+        ));
+    }
+    if username.is_empty() != password.is_empty() {
+        return Err(TargetError::Configuration(
+            "Pulsar username and password must be specified together".to_string(),
+        ));
+    }
+
+    let tls_ca = config.lookup(PULSAR_TLS_CA).unwrap_or_default();
+    let tls_allow_insecure = parse_target_bool(config.lookup(PULSAR_TLS_ALLOW_INSECURE).as_deref()).unwrap_or(false);
+    let tls_hostname_verification = parse_target_bool(config.lookup(PULSAR_TLS_HOSTNAME_VERIFICATION).as_deref()).unwrap_or(true);
+
+    if !tls_ca.is_empty() && !Path::new(&tls_ca).is_absolute() {
+        return Err(TargetError::Configuration("Pulsar tls_ca must be an absolute path".to_string()));
+    }
+    if url.scheme() != "pulsar+ssl" && (!tls_ca.is_empty() || tls_allow_insecure || !tls_hostname_verification) {
+        return Err(TargetError::Configuration(
+            "Pulsar TLS settings are only allowed with pulsar+ssl brokers".to_string(),
+        ));
+    }
+
+    let queue_dir = config
+        .lookup(PULSAR_QUEUE_DIR)
+        .unwrap_or_else(|| default_queue_dir.to_string());
+    if !queue_dir.is_empty() && !Path::new(&queue_dir).is_absolute() {
+        return Err(TargetError::Configuration("Pulsar queue directory must be an absolute path".to_string()));
+    }
+
+    Ok(())
+}
+
+pub(super) fn parse_url(value: &str, field_label: &str) -> Result<Url, TargetError> {
+    Url::parse(value).map_err(|e| TargetError::Configuration(format!("Invalid {field_label}: {e} (value: '{value}')")))
+}

crates/targets/src/config/loader.rs

@@ -0,0 +1,239 @@
+// 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 super::common::{is_target_enabled, split_env_field_and_instance};
+use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, EnableState};
+use rustfs_ecstore::config::{Config, KVS};
+use std::collections::{HashMap, HashSet};
+use std::str::FromStr;
+use tracing::{debug, warn};
+
+pub fn collect_target_configs(
+    config: &Config,
+    route_prefix: &str,
+    target_type: &str,
+    valid_fields: &HashSet<String>,
+) -> Vec<(String, KVS)> {
+    collect_target_configs_from_env(config, route_prefix, target_type, valid_fields, std::env::vars())
+}
+
+pub fn collect_env_target_instance_ids(route_prefix: &str, target_type: &str, valid_fields: &HashSet<String>) -> HashSet<String> {
+    collect_env_target_instance_ids_from_env(route_prefix, target_type, valid_fields, std::env::vars())
+}
+
+pub fn collect_env_target_instance_ids_from_env<I>(
+    route_prefix: &str,
+    target_type: &str,
+    valid_fields: &HashSet<String>,
+    env_vars: I,
+) -> HashSet<String>
+where
+    I: IntoIterator<Item = (String, String)>,
+{
+    let env_prefix = format!("{ENV_PREFIX}{route_prefix}{target_type}{DEFAULT_DELIMITER}").to_uppercase();
+    let mut instance_ids = HashSet::new();
+
+    for (key, _value) in env_vars.into_iter().filter(|(key, _)| key.starts_with(ENV_PREFIX)) {
+        let Some(rest) = key.strip_prefix(&env_prefix) else {
+            continue;
+        };
+        let Some((_field_name, instance_id)) = split_env_field_and_instance(rest, valid_fields) else {
+            continue;
+        };
+        if instance_id != DEFAULT_DELIMITER && !instance_id.is_empty() {
+            instance_ids.insert(instance_id);
+        }
+    }
+
+    instance_ids
+}
+
+pub fn collect_target_configs_from_env<I>(
+    config: &Config,
+    route_prefix: &str,
+    target_type: &str,
+    valid_fields: &HashSet<String>,
+    env_vars: I,
+) -> Vec<(String, KVS)>
+where
+    I: IntoIterator<Item = (String, String)>,
+{
+    let all_env: Vec<(String, String)> = env_vars.into_iter().filter(|(key, _)| key.starts_with(ENV_PREFIX)).collect();
+    let section_name = format!("{route_prefix}{target_type}").to_lowercase();
+    let file_configs = config.0.get(&section_name).cloned().unwrap_or_default();
+    let default_cfg = file_configs.get(DEFAULT_DELIMITER).cloned().unwrap_or_default();
+
+    let enable_prefix =
+        format!("{ENV_PREFIX}{route_prefix}{target_type}{DEFAULT_DELIMITER}{ENABLE_KEY}{DEFAULT_DELIMITER}").to_uppercase();
+    let env_prefix = format!("{ENV_PREFIX}{route_prefix}{target_type}{DEFAULT_DELIMITER}").to_uppercase();
+
+    let mut instance_ids_from_env = HashSet::new();
+    let mut env_overrides: HashMap<String, KVS> = HashMap::new();
+    for (key, value) in &all_env {
+        if EnableState::from_str(value).ok().map(|s| s.is_enabled()).unwrap_or(false)
+            && let Some(id) = key.strip_prefix(&enable_prefix)
+            && !id.is_empty()
+        {
+            instance_ids_from_env.insert(id.to_lowercase());
+        }
+
+        let Some(rest) = key.strip_prefix(&env_prefix) else {
+            continue;
+        };
+
+        let Some((field_name, instance_id)) = split_env_field_and_instance(rest, valid_fields) else {
+            warn!(
+                field_name = %rest.to_lowercase(),
+                "Ignore environment variable field not found in the valid field list for target type {}",
+                target_type
+            );
+            continue;
+        };
+
+        debug!(
+            instance_id = %if instance_id == DEFAULT_DELIMITER { DEFAULT_DELIMITER } else { &instance_id },
+            %field_name,
+            %value,
+            "Parsed target environment override"
+        );
+        env_overrides
+            .entry(instance_id)
+            .or_default()
+            .insert(field_name, value.clone());
+    }
+
+    let mut effective_default = default_cfg;
+    if let Some(default_env_cfg) = env_overrides.remove(DEFAULT_DELIMITER) {
+        effective_default.extend(default_env_cfg);
+    }
+
+    let mut all_instance_ids: Vec<String> = file_configs
+        .keys()
+        .filter(|key| key.as_str() != DEFAULT_DELIMITER)
+        .cloned()
+        .collect();
+    all_instance_ids.extend(instance_ids_from_env);
+    all_instance_ids.sort();
+    all_instance_ids.dedup();
+
+    let mut merged_configs = Vec::new();
+    for id in all_instance_ids {
+        let mut merged_config = effective_default.clone();
+        if let Some(file_instance_cfg) = file_configs.get(&id) {
+            merged_config.extend(file_instance_cfg.clone());
+        }
+        if let Some(env_instance_cfg) = env_overrides.get(&id) {
+            merged_config.extend(env_instance_cfg.clone());
+        }
+
+        debug!(instance_id = %id, ?merged_config, "Merged target configuration");
+        if is_target_enabled(&merged_config) {
+            merged_configs.push((id, merged_config));
+        }
+    }
+
+    merged_configs
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{collect_env_target_instance_ids_from_env, collect_target_configs_from_env};
+    use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
+    use rustfs_config::{ENABLE_KEY, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_LIMIT};
+    use rustfs_ecstore::config::{Config, KVS};
+    use std::collections::{HashMap, HashSet};
+
+    #[test]
+    fn collect_target_configs_applies_default_env_overrides_to_file_targets() {
+        let mut cfg = Config(HashMap::new());
+        let mut subsystem = HashMap::new();
+
+        let mut default_kvs = KVS::new();
+        default_kvs.insert(ENABLE_KEY.to_string(), "off".to_string());
+        subsystem.insert("_".to_string(), default_kvs);
+
+        let mut primary = KVS::new();
+        primary.insert(WEBHOOK_ENDPOINT.to_string(), "https://example.com/primary".to_string());
+        subsystem.insert("primary".to_string(), primary);
+
+        let mut secondary = KVS::new();
+        secondary.insert(WEBHOOK_ENDPOINT.to_string(), "https://example.com/secondary".to_string());
+        subsystem.insert("secondary".to_string(), secondary);
+
+        cfg.0.insert("notify_webhook".to_string(), subsystem);
+
+        let configs = collect_target_configs_from_env(
+            &cfg,
+            NOTIFY_ROUTE_PREFIX,
+            "webhook",
+            &HashSet::from([
+                ENABLE_KEY.to_string(),
+                WEBHOOK_ENDPOINT.to_string(),
+                WEBHOOK_QUEUE_LIMIT.to_string(),
+            ]),
+            vec![
+                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE".to_string(), "on".to_string()),
+                ("RUSTFS_NOTIFY_WEBHOOK_QUEUE_LIMIT".to_string(), "42".to_string()),
+            ],
+        );
+
+        let configs: HashMap<String, KVS> = configs.into_iter().collect();
+        assert_eq!(configs.len(), 2);
+        assert_eq!(configs["primary"].lookup(ENABLE_KEY).as_deref(), Some("on"));
+        assert_eq!(configs["secondary"].lookup(ENABLE_KEY).as_deref(), Some("on"));
+        assert_eq!(configs["primary"].lookup(WEBHOOK_QUEUE_LIMIT).as_deref(), Some("42"));
+        assert_eq!(configs["secondary"].lookup(WEBHOOK_QUEUE_LIMIT).as_deref(), Some("42"));
+    }
+
+    #[test]
+    fn collect_target_configs_discovers_enabled_instance_from_env() {
+        let cfg = Config(HashMap::new());
+        let configs = collect_target_configs_from_env(
+            &cfg,
+            NOTIFY_ROUTE_PREFIX,
+            "webhook",
+            &HashSet::from([ENABLE_KEY.to_string(), WEBHOOK_ENDPOINT.to_string()]),
+            vec![
+                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(), "on".to_string()),
+                (
+                    "RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY".to_string(),
+                    "https://example.com/from-env".to_string(),
+                ),
+            ],
+        );
+
+        assert_eq!(configs.len(), 1);
+        assert_eq!(configs[0].0, "primary");
+        assert_eq!(configs[0].1.lookup(WEBHOOK_ENDPOINT).as_deref(), Some("https://example.com/from-env"));
+    }
+
+    #[test]
+    fn collect_env_target_instance_ids_handles_keys_with_internal_underscores() {
+        let ids = collect_env_target_instance_ids_from_env(
+            NOTIFY_ROUTE_PREFIX,
+            "webhook",
+            &HashSet::from([
+                ENABLE_KEY.to_string(),
+                WEBHOOK_ENDPOINT.to_string(),
+                WEBHOOK_QUEUE_LIMIT.to_string(),
+            ]),
+            vec![
+                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(), "on".to_string()),
+                ("RUSTFS_NOTIFY_WEBHOOK_QUEUE_LIMIT_PRIMARY".to_string(), "42".to_string()),
+            ],
+        );
+
+        assert_eq!(ids, HashSet::from(["primary".to_string()]));
+    }
+}

crates/targets/src/config/mod.rs

@@ -0,0 +1,26 @@
+// 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.
+
+mod common;
+mod loader;
+mod target_args;
+
+pub use loader::{
+    collect_env_target_instance_ids, collect_env_target_instance_ids_from_env, collect_target_configs,
+    collect_target_configs_from_env,
+};
+pub use target_args::{
+    build_mqtt_args, build_nats_args, build_pulsar_args, build_webhook_args, validate_mqtt_config, validate_nats_config,
+    validate_pulsar_config, validate_webhook_config,
+};

crates/targets/src/config/target_args.rs

@@ -0,0 +1,267 @@
+// 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 super::common::{parse_target_bool, parse_url, validate_nats_server_config, validate_pulsar_broker_config};
+use crate::error::TargetError;
+use crate::target::{
+    TargetType,
+    mqtt::{MQTTArgs, MQTTTlsConfig, validate_mqtt_broker_url},
+    nats::{NATSArgs, validate_nats_address},
+    pulsar::{PulsarArgs, validate_pulsar_broker},
+    webhook::WebhookArgs,
+};
+use rumqttc::QoS;
+use rustfs_config::{
+    DEFAULT_LIMIT, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT,
+    MQTT_RECONNECT_INTERVAL, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_POLICY, MQTT_TLS_TRUST_LEAF_AS_CA,
+    MQTT_TOPIC, MQTT_USERNAME, MQTT_WS_PATH_ALLOWLIST, NATS_ADDRESS, NATS_CREDENTIALS_FILE, NATS_PASSWORD, NATS_QUEUE_DIR,
+    NATS_QUEUE_LIMIT, NATS_SUBJECT, NATS_TLS_CA, NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY, NATS_TLS_REQUIRED, NATS_TOKEN,
+    NATS_USERNAME, PULSAR_AUTH_TOKEN, PULSAR_BROKER, PULSAR_PASSWORD, PULSAR_QUEUE_DIR, PULSAR_QUEUE_LIMIT,
+    PULSAR_TLS_ALLOW_INSECURE, PULSAR_TLS_CA, PULSAR_TLS_HOSTNAME_VERIFICATION, PULSAR_TOPIC, PULSAR_USERNAME,
+    RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT, WEBHOOK_AUTH_TOKEN, WEBHOOK_CLIENT_CA, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY,
+    WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_SKIP_TLS_VERIFY,
+};
+use rustfs_ecstore::config::KVS;
+use std::path::Path;
+use std::time::Duration;
+
+pub fn build_webhook_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<WebhookArgs, TargetError> {
+    let endpoint = config
+        .lookup(WEBHOOK_ENDPOINT)
+        .ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
+    let parsed_endpoint = endpoint.trim();
+    let endpoint_url = parse_url(parsed_endpoint, "endpoint URL")?;
+
+    Ok(WebhookArgs {
+        enable: true,
+        endpoint: endpoint_url,
+        auth_token: config.lookup(WEBHOOK_AUTH_TOKEN).unwrap_or_default(),
+        queue_dir: config
+            .lookup(WEBHOOK_QUEUE_DIR)
+            .unwrap_or_else(|| default_queue_dir.to_string()),
+        queue_limit: config
+            .lookup(WEBHOOK_QUEUE_LIMIT)
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(DEFAULT_LIMIT),
+        client_cert: config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default(),
+        client_key: config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default(),
+        client_ca: config.lookup(WEBHOOK_CLIENT_CA).unwrap_or_default(),
+        skip_tls_verify: config
+            .lookup(WEBHOOK_SKIP_TLS_VERIFY)
+            .and_then(|v| v.parse::<bool>().ok())
+            .unwrap_or(RUSTFS_WEBHOOK_SKIP_TLS_VERIFY_DEFAULT),
+        target_type,
+    })
+}
+
+pub fn validate_webhook_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
+    let endpoint = config
+        .lookup(WEBHOOK_ENDPOINT)
+        .ok_or_else(|| TargetError::Configuration("Missing webhook endpoint".to_string()))?;
+    let parsed_endpoint = endpoint.trim();
+    let _ = parse_url(parsed_endpoint, "endpoint URL")?;
+
+    let client_cert = config.lookup(WEBHOOK_CLIENT_CERT).unwrap_or_default();
+    let client_key = config.lookup(WEBHOOK_CLIENT_KEY).unwrap_or_default();
+    if client_cert.is_empty() != client_key.is_empty() {
+        return Err(TargetError::Configuration(
+            "Both client_cert and client_key must be specified together".to_string(),
+        ));
+    }
+
+    let queue_dir = config
+        .lookup(WEBHOOK_QUEUE_DIR)
+        .unwrap_or_else(|| default_queue_dir.to_string());
+    if !queue_dir.is_empty() && !Path::new(&queue_dir).is_absolute() {
+        return Err(TargetError::Configuration("Webhook queue directory must be an absolute path".to_string()));
+    }
+
+    Ok(())
+}
+
+pub fn build_mqtt_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<MQTTArgs, TargetError> {
+    let broker = config
+        .lookup(MQTT_BROKER)
+        .ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
+    let broker_url = parse_url(&broker, "broker URL")?;
+
+    let topic = config
+        .lookup(MQTT_TOPIC)
+        .ok_or_else(|| TargetError::Configuration("Missing MQTT topic".to_string()))?;
+
+    Ok(MQTTArgs {
+        enable: true,
+        broker: broker_url,
+        topic,
+        qos: config
+            .lookup(MQTT_QOS)
+            .and_then(|v| v.parse::<u8>().ok())
+            .map(|q| match q {
+                0 => QoS::AtMostOnce,
+                1 => QoS::AtLeastOnce,
+                2 => QoS::ExactlyOnce,
+                _ => QoS::AtLeastOnce,
+            })
+            .unwrap_or(QoS::AtLeastOnce),
+        username: config.lookup(MQTT_USERNAME).unwrap_or_default(),
+        password: config.lookup(MQTT_PASSWORD).unwrap_or_default(),
+        max_reconnect_interval: config
+            .lookup(MQTT_RECONNECT_INTERVAL)
+            .and_then(|v| v.parse::<u64>().ok())
+            .map(Duration::from_secs)
+            .unwrap_or_else(|| Duration::from_secs(5)),
+        keep_alive: config
+            .lookup(MQTT_KEEP_ALIVE_INTERVAL)
+            .and_then(|v| v.parse::<u64>().ok())
+            .map(Duration::from_secs)
+            .unwrap_or_else(|| Duration::from_secs(30)),
+        tls: MQTTTlsConfig::from_values(
+            config.lookup(MQTT_TLS_POLICY).as_deref(),
+            config.lookup(MQTT_TLS_CA).as_deref(),
+            config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
+            config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
+            config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
+            config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
+        )?,
+        queue_dir: config.lookup(MQTT_QUEUE_DIR).unwrap_or_else(|| default_queue_dir.to_string()),
+        queue_limit: config
+            .lookup(MQTT_QUEUE_LIMIT)
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(DEFAULT_LIMIT),
+        target_type,
+    })
+}
+
+pub fn validate_mqtt_config(config: &KVS) -> Result<(), TargetError> {
+    let broker = config
+        .lookup(MQTT_BROKER)
+        .ok_or_else(|| TargetError::Configuration("Missing MQTT broker".to_string()))?;
+    let url = parse_url(&broker, "broker URL")?;
+
+    let tls = MQTTTlsConfig::from_values(
+        config.lookup(MQTT_TLS_POLICY).as_deref(),
+        config.lookup(MQTT_TLS_CA).as_deref(),
+        config.lookup(MQTT_TLS_CLIENT_CERT).as_deref(),
+        config.lookup(MQTT_TLS_CLIENT_KEY).as_deref(),
+        config.lookup(MQTT_TLS_TRUST_LEAF_AS_CA).as_deref(),
+        config.lookup(MQTT_WS_PATH_ALLOWLIST).as_deref(),
+    )?;
+    validate_mqtt_broker_url(&url, &tls)?;
+
+    if config.lookup(MQTT_TOPIC).is_none() {
+        return Err(TargetError::Configuration("Missing MQTT topic".to_string()));
+    }
+
+    if let Some(qos_str) = config.lookup(MQTT_QOS) {
+        let qos = qos_str
+            .parse::<u8>()
+            .map_err(|_| TargetError::Configuration("Invalid QoS value".to_string()))?;
+        if qos > 2 {
+            return Err(TargetError::Configuration("QoS must be 0, 1, or 2".to_string()));
+        }
+    }
+
+    let queue_dir = config.lookup(MQTT_QUEUE_DIR).unwrap_or_default();
+    if !queue_dir.is_empty() {
+        if !Path::new(&queue_dir).is_absolute() {
+            return Err(TargetError::Configuration("MQTT queue directory must be an absolute path".to_string()));
+        }
+        if let Some(qos_str) = config.lookup(MQTT_QOS)
+            && qos_str == "0"
+        {
+            return Err(TargetError::Configuration(
+                "QoS should be AtLeastOnce (1) or ExactlyOnce (2) if queue_dir is set".to_string(),
+            ));
+        }
+    }
+
+    Ok(())
+}
+
+pub fn build_nats_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<NATSArgs, TargetError> {
+    let address = config
+        .lookup(NATS_ADDRESS)
+        .ok_or_else(|| TargetError::Configuration("Missing NATS address".to_string()))?;
+    validate_nats_address(&address)?;
+
+    let subject = config
+        .lookup(NATS_SUBJECT)
+        .ok_or_else(|| TargetError::Configuration("Missing NATS subject".to_string()))?;
+
+    Ok(NATSArgs {
+        enable: true,
+        address,
+        subject,
+        username: config.lookup(NATS_USERNAME).unwrap_or_default(),
+        password: config.lookup(NATS_PASSWORD).unwrap_or_default(),
+        token: config.lookup(NATS_TOKEN).unwrap_or_default(),
+        credentials_file: config.lookup(NATS_CREDENTIALS_FILE).unwrap_or_default(),
+        tls_ca: config.lookup(NATS_TLS_CA).unwrap_or_default(),
+        tls_client_cert: config.lookup(NATS_TLS_CLIENT_CERT).unwrap_or_default(),
+        tls_client_key: config.lookup(NATS_TLS_CLIENT_KEY).unwrap_or_default(),
+        tls_required: parse_target_bool(config.lookup(NATS_TLS_REQUIRED).as_deref()).unwrap_or(false),
+        queue_dir: config.lookup(NATS_QUEUE_DIR).unwrap_or_else(|| default_queue_dir.to_string()),
+        queue_limit: config
+            .lookup(NATS_QUEUE_LIMIT)
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(DEFAULT_LIMIT),
+        target_type,
+    })
+}
+
+pub fn validate_nats_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
+    let address = config
+        .lookup(NATS_ADDRESS)
+        .ok_or_else(|| TargetError::Configuration("Missing NATS address".to_string()))?;
+    let server = validate_nats_address(&address)?;
+    validate_nats_server_config(&server, config, default_queue_dir)
+}
+
+pub fn build_pulsar_args(config: &KVS, default_queue_dir: &str, target_type: TargetType) -> Result<PulsarArgs, TargetError> {
+    let broker = config
+        .lookup(PULSAR_BROKER)
+        .ok_or_else(|| TargetError::Configuration("Missing Pulsar broker".to_string()))?;
+    validate_pulsar_broker(&broker)?;
+
+    let topic = config
+        .lookup(PULSAR_TOPIC)
+        .ok_or_else(|| TargetError::Configuration("Missing Pulsar topic".to_string()))?;
+
+    Ok(PulsarArgs {
+        enable: true,
+        broker,
+        topic,
+        auth_token: config.lookup(PULSAR_AUTH_TOKEN).unwrap_or_default(),
+        username: config.lookup(PULSAR_USERNAME).unwrap_or_default(),
+        password: config.lookup(PULSAR_PASSWORD).unwrap_or_default(),
+        tls_ca: config.lookup(PULSAR_TLS_CA).unwrap_or_default(),
+        tls_allow_insecure: parse_target_bool(config.lookup(PULSAR_TLS_ALLOW_INSECURE).as_deref()).unwrap_or(false),
+        tls_hostname_verification: parse_target_bool(config.lookup(PULSAR_TLS_HOSTNAME_VERIFICATION).as_deref()).unwrap_or(true),
+        queue_dir: config
+            .lookup(PULSAR_QUEUE_DIR)
+            .unwrap_or_else(|| default_queue_dir.to_string()),
+        queue_limit: config
+            .lookup(PULSAR_QUEUE_LIMIT)
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(DEFAULT_LIMIT),
+        target_type,
+    })
+}
+
+pub fn validate_pulsar_config(config: &KVS, default_queue_dir: &str) -> Result<(), TargetError> {
+    let broker = config
+        .lookup(PULSAR_BROKER)
+        .ok_or_else(|| TargetError::Configuration("Missing Pulsar broker".to_string()))?;
+    validate_pulsar_broker_config(&broker, config, default_queue_dir)
+}

crates/targets/src/lib.rs

@@ -14,12 +14,15 @@
 
 pub mod arn;
 mod check;
+pub mod config;
 pub mod error;
 pub mod store;
 pub mod sys;
 pub mod target;
 
-pub use check::{check_mqtt_broker_available, check_mqtt_broker_available_with_tls};
+pub use check::{
+    check_mqtt_broker_available, check_mqtt_broker_available_with_tls, check_nats_server_available, check_pulsar_broker_available,
+};
 pub use error::{StoreError, TargetError};
 pub use rustfs_s3_common::EventName;
 use serde::{Deserialize, Serialize};

crates/targets/src/target/mod.rs

@@ -26,6 +26,8 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use tracing::warn;
 
 pub mod mqtt;
+pub mod nats;
+pub mod pulsar;
 pub mod webhook;
 
 /// A read-only snapshot of delivery counters for a target.
@@ -286,6 +288,8 @@ pub enum ChannelTargetType {
     Webhook,
     Kafka,
     Mqtt,
+    Nats,
+    Pulsar,
 }
 
 impl ChannelTargetType {
@@ -294,6 +298,8 @@ impl ChannelTargetType {
             ChannelTargetType::Webhook => "webhook",
             ChannelTargetType::Kafka => "kafka",
             ChannelTargetType::Mqtt => "mqtt",
+            ChannelTargetType::Nats => "nats",
+            ChannelTargetType::Pulsar => "pulsar",
         }
     }
 }
@@ -304,6 +310,8 @@ impl std::fmt::Display for ChannelTargetType {
             ChannelTargetType::Webhook => write!(f, "webhook"),
             ChannelTargetType::Kafka => write!(f, "kafka"),
             ChannelTargetType::Mqtt => write!(f, "mqtt"),
+            ChannelTargetType::Nats => write!(f, "nats"),
+            ChannelTargetType::Pulsar => write!(f, "pulsar"),
         }
     }
 }

crates/targets/src/target/nats.rs

@@ -0,0 +1,367 @@
+// 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::{
+    StoreError, Target, TargetLog,
+    arn::TargetID,
+    error::TargetError,
+    store::{Key, QueueStore, Store},
+    target::{
+        ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
+        TargetType,
+    },
+};
+use async_trait::async_trait;
+use rustfs_config::{NATS_CREDENTIALS_FILE, NATS_TLS_CA, NATS_TLS_CLIENT_CERT, NATS_TLS_CLIENT_KEY};
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use std::path::{Path, PathBuf};
+use std::str::FromStr;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use tracing::{error, info, instrument};
+
+#[derive(Debug, Clone)]
+pub struct NATSArgs {
+    pub enable: bool,
+    pub address: String,
+    pub subject: String,
+    pub username: String,
+    pub password: String,
+    pub token: String,
+    pub credentials_file: String,
+    pub tls_ca: String,
+    pub tls_client_cert: String,
+    pub tls_client_key: String,
+    pub tls_required: bool,
+    pub queue_dir: String,
+    pub queue_limit: u64,
+    pub target_type: TargetType,
+}
+
+impl NATSArgs {
+    pub fn validate(&self) -> Result<(), TargetError> {
+        if !self.enable {
+            return Ok(());
+        }
+
+        validate_nats_address(&self.address)?;
+        validate_nats_auth(self)?;
+
+        if self.subject.trim().is_empty() || self.subject.chars().any(char::is_whitespace) {
+            return Err(TargetError::Configuration(
+                "NATS subject cannot be empty or contain whitespace".to_string(),
+            ));
+        }
+
+        if !self.credentials_file.is_empty() && !Path::new(&self.credentials_file).is_absolute() {
+            return Err(TargetError::Configuration(format!("{NATS_CREDENTIALS_FILE} must be an absolute path")));
+        }
+        if !self.tls_ca.is_empty() && !Path::new(&self.tls_ca).is_absolute() {
+            return Err(TargetError::Configuration(format!("{NATS_TLS_CA} must be an absolute path")));
+        }
+        if !self.tls_client_cert.is_empty() && !Path::new(&self.tls_client_cert).is_absolute() {
+            return Err(TargetError::Configuration(format!("{NATS_TLS_CLIENT_CERT} must be an absolute path")));
+        }
+        if !self.tls_client_key.is_empty() && !Path::new(&self.tls_client_key).is_absolute() {
+            return Err(TargetError::Configuration(format!("{NATS_TLS_CLIENT_KEY} must be an absolute path")));
+        }
+        if self.tls_client_cert.is_empty() != self.tls_client_key.is_empty() {
+            return Err(TargetError::Configuration(
+                "NATS tls_client_cert and tls_client_key must be specified together".to_string(),
+            ));
+        }
+
+        if !self.queue_dir.is_empty() && !Path::new(&self.queue_dir).is_absolute() {
+            return Err(TargetError::Configuration("NATS queue directory must be an absolute path".to_string()));
+        }
+
+        Ok(())
+    }
+}
+
+pub fn validate_nats_address(address: &str) -> Result<async_nats::ServerAddr, TargetError> {
+    let server = async_nats::ServerAddr::from_str(address)
+        .map_err(|e| TargetError::Configuration(format!("Invalid NATS address: {e}")))?;
+
+    if server.has_user_pass() {
+        return Err(TargetError::Configuration("NATS address must not embed username or password".to_string()));
+    }
+
+    Ok(server)
+}
+
+fn validate_nats_auth(args: &NATSArgs) -> Result<(), TargetError> {
+    let mut auth_methods = 0usize;
+
+    if !args.token.is_empty() {
+        auth_methods += 1;
+    }
+
+    if !args.credentials_file.is_empty() {
+        auth_methods += 1;
+    }
+
+    let has_user = !args.username.is_empty();
+    let has_password = !args.password.is_empty();
+    if has_user || has_password {
+        if has_user != has_password {
+            return Err(TargetError::Configuration(
+                "NATS username and password must be specified together".to_string(),
+            ));
+        }
+        auth_methods += 1;
+    }
+
+    if auth_methods > 1 {
+        return Err(TargetError::Configuration(
+            "NATS supports only one auth method at a time: token, username/password, or credentials_file".to_string(),
+        ));
+    }
+
+    Ok(())
+}
+
+pub async fn connect_nats(args: &NATSArgs) -> Result<async_nats::Client, TargetError> {
+    args.validate()?;
+
+    let mut options = async_nats::ConnectOptions::new().require_tls(args.tls_required);
+
+    if !args.token.is_empty() {
+        options = options.token(args.token.clone());
+    } else if !args.username.is_empty() {
+        options = options.user_and_password(args.username.clone(), args.password.clone());
+    } else if !args.credentials_file.is_empty() {
+        options = options
+            .credentials_file(&args.credentials_file)
+            .await
+            .map_err(|e| TargetError::Configuration(format!("Failed to load NATS credentials file: {e}")))?;
+    }
+
+    if !args.tls_ca.is_empty() {
+        options = options.add_root_certificates(PathBuf::from(&args.tls_ca));
+    }
+    if !args.tls_client_cert.is_empty() {
+        options = options.add_client_certificate(PathBuf::from(&args.tls_client_cert), PathBuf::from(&args.tls_client_key));
+    }
+
+    options
+        .connect(args.address.clone())
+        .await
+        .map_err(|e| TargetError::Network(format!("Failed to connect to NATS server: {e}")))
+}
+
+pub struct NATSTarget<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    id: TargetID,
+    args: NATSArgs,
+    client: Mutex<Option<async_nats::Client>>,
+    store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
+    connected: AtomicBool,
+    delivery_counters: Arc<TargetDeliveryCounters>,
+    _phantom: std::marker::PhantomData<E>,
+}
+
+impl<E> NATSTarget<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    pub fn clone_box(&self) -> Box<dyn Target<E> + Send + Sync> {
+        Box::new(NATSTarget::<E> {
+            id: self.id.clone(),
+            args: self.args.clone(),
+            client: Mutex::new(self.client.lock().unwrap().clone()),
+            store: self.store.as_ref().map(|s| s.boxed_clone()),
+            connected: AtomicBool::new(self.connected.load(Ordering::SeqCst)),
+            delivery_counters: Arc::clone(&self.delivery_counters),
+            _phantom: std::marker::PhantomData,
+        })
+    }
+
+    #[instrument(skip(args), fields(target_id_as_string = %id))]
+    pub fn new(id: String, args: NATSArgs) -> Result<Self, TargetError> {
+        args.validate()?;
+        let target_id = TargetID::new(id, ChannelTargetType::Nats.as_str().to_string());
+        let queue_store = if !args.queue_dir.is_empty() {
+            let base_path = PathBuf::from(&args.queue_dir);
+            let specific_queue_path = base_path.join(format!("rustfs-{}-{}", ChannelTargetType::Nats.as_str(), target_id.id));
+            let extension = match args.target_type {
+                TargetType::AuditLog => rustfs_config::audit::AUDIT_STORE_EXTENSION,
+                TargetType::NotifyEvent => rustfs_config::notify::NOTIFY_STORE_EXTENSION,
+            };
+            let store = QueueStore::<QueuedPayload>::new(specific_queue_path, args.queue_limit, extension);
+            if let Err(e) = store.open() {
+                error!(target_id = %target_id, error = %e, "Failed to open store for NATS target");
+                return Err(TargetError::Storage(format!("{e}")));
+            }
+            Some(Box::new(store) as Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>)
+        } else {
+            None
+        };
+
+        Ok(Self {
+            id: target_id,
+            args,
+            client: Mutex::new(None),
+            store: queue_store,
+            connected: AtomicBool::new(false),
+            delivery_counters: Arc::new(TargetDeliveryCounters::default()),
+            _phantom: std::marker::PhantomData,
+        })
+    }
+
+    async fn get_or_connect(&self) -> Result<async_nats::Client, TargetError> {
+        if let Some(client) = self.client.lock().unwrap().clone() {
+            return Ok(client);
+        }
+
+        let client = connect_nats(&self.args).await?;
+        client
+            .flush()
+            .await
+            .map_err(|e| TargetError::Network(format!("Failed to flush NATS connection: {e}")))?;
+        self.connected.store(true, Ordering::SeqCst);
+
+        let mut guard = self.client.lock().unwrap();
+        let shared = guard.get_or_insert_with(|| client.clone()).clone();
+        Ok(shared)
+    }
+
+    fn build_queued_payload(&self, event: &EntityTarget<E>) -> Result<QueuedPayload, TargetError> {
+        let object_name = crate::target::decode_object_name(&event.object_name)?;
+        let key = format!("{}/{}", event.bucket_name, object_name);
+        let log = TargetLog {
+            event_name: event.event_name,
+            key,
+            records: vec![event.clone()],
+        };
+        let body = serde_json::to_vec(&log).map_err(|e| TargetError::Serialization(format!("Failed to serialize event: {e}")))?;
+        let meta = QueuedPayloadMeta::new(
+            event.event_name,
+            event.bucket_name.clone(),
+            event.object_name.clone(),
+            "application/json",
+            body.len(),
+        );
+        Ok(QueuedPayload::new(meta, body))
+    }
+
+    async fn send_body(&self, body: Vec<u8>) -> Result<(), TargetError> {
+        let client = self.get_or_connect().await?;
+        client
+            .publish(self.args.subject.clone(), body.into())
+            .await
+            .map_err(|e| TargetError::Request(format!("Failed to publish NATS message: {e}")))?;
+        self.delivery_counters.record_success();
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl<E> Target<E> for NATSTarget<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    fn id(&self) -> TargetID {
+        self.id.clone()
+    }
+
+    async fn is_active(&self) -> Result<bool, TargetError> {
+        let client = self.get_or_connect().await?;
+        client
+            .flush()
+            .await
+            .map_err(|e| TargetError::Network(format!("NATS health check failed: {e}")))?;
+        Ok(true)
+    }
+
+    async fn save(&self, event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
+        let queued = match self.build_queued_payload(&event) {
+            Ok(queued) => queued,
+            Err(err) => {
+                self.delivery_counters.record_final_failure();
+                return Err(err);
+            }
+        };
+
+        if let Some(store) = &self.store {
+            let encoded = match queued.encode() {
+                Ok(encoded) => encoded,
+                Err(err) => {
+                    self.delivery_counters.record_final_failure();
+                    return Err(TargetError::Storage(format!("Failed to encode queued payload: {err}")));
+                }
+            };
+            if let Err(e) = store.put_raw(&encoded) {
+                self.delivery_counters.record_final_failure();
+                return Err(TargetError::Storage(format!("Failed to save event to store: {e}")));
+            }
+            Ok(())
+        } else {
+            if let Err(err) = self.send_body(queued.body).await {
+                self.delivery_counters.record_final_failure();
+                return Err(err);
+            }
+            Ok(())
+        }
+    }
+
+    async fn send_raw_from_store(&self, _key: Key, body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
+        self.send_body(body).await
+    }
+
+    async fn close(&self) -> Result<(), TargetError> {
+        let client = self.client.lock().unwrap().take();
+        self.connected.store(false, Ordering::SeqCst);
+        if let Some(client) = client {
+            client
+                .drain()
+                .await
+                .map_err(|e| TargetError::Network(format!("Failed to drain NATS client: {e}")))?;
+        }
+        info!(target_id = %self.id, "NATS target closed");
+        Ok(())
+    }
+
+    fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
+        self.store.as_deref()
+    }
+
+    fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
+        self.clone_box()
+    }
+
+    async fn init(&self) -> Result<(), TargetError> {
+        if !self.is_enabled() {
+            return Ok(());
+        }
+        let _ = self.get_or_connect().await?;
+        Ok(())
+    }
+
+    fn is_enabled(&self) -> bool {
+        self.args.enable
+    }
+
+    fn delivery_snapshot(&self) -> TargetDeliverySnapshot {
+        self.delivery_counters
+            .snapshot(self.store.as_deref().map_or(0, |store| store.len() as u64))
+    }
+
+    fn record_final_failure(&self) {
+        self.delivery_counters.record_final_failure();
+    }
+}

crates/targets/src/target/pulsar.rs

@@ -0,0 +1,387 @@
+// 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::{
+    StoreError, Target, TargetLog,
+    arn::TargetID,
+    error::TargetError,
+    store::{Key, QueueStore, Store},
+    target::{
+        ChannelTargetType, EntityTarget, QueuedPayload, QueuedPayloadMeta, TargetDeliveryCounters, TargetDeliverySnapshot,
+        TargetType,
+    },
+};
+use async_trait::async_trait;
+use pulsar::{Authentication, Producer, Pulsar, TokioExecutor};
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use tokio::sync::Mutex as AsyncMutex;
+use tracing::{error, info, instrument};
+use url::Url;
+
+#[derive(Debug, Clone)]
+pub struct PulsarArgs {
+    pub enable: bool,
+    pub broker: String,
+    pub topic: String,
+    pub auth_token: String,
+    pub username: String,
+    pub password: String,
+    pub tls_ca: String,
+    pub tls_allow_insecure: bool,
+    pub tls_hostname_verification: bool,
+    pub queue_dir: String,
+    pub queue_limit: u64,
+    pub target_type: TargetType,
+}
+
+impl PulsarArgs {
+    pub fn validate(&self) -> Result<(), TargetError> {
+        if !self.enable {
+            return Ok(());
+        }
+
+        validate_pulsar_broker(&self.broker)?;
+
+        if self.topic.trim().is_empty() {
+            return Err(TargetError::Configuration("Pulsar topic cannot be empty".to_string()));
+        }
+
+        if !self.auth_token.is_empty() && (!self.username.is_empty() || !self.password.is_empty()) {
+            return Err(TargetError::Configuration(
+                "Pulsar supports either auth_token or username/password auth, not both".to_string(),
+            ));
+        }
+
+        if self.username.is_empty() != self.password.is_empty() {
+            return Err(TargetError::Configuration(
+                "Pulsar username and password must be specified together".to_string(),
+            ));
+        }
+
+        if !self.tls_ca.is_empty() && !Path::new(&self.tls_ca).is_absolute() {
+            return Err(TargetError::Configuration("Pulsar tls_ca must be an absolute path".to_string()));
+        }
+
+        if !self.queue_dir.is_empty() && !Path::new(&self.queue_dir).is_absolute() {
+            return Err(TargetError::Configuration("Pulsar queue directory must be an absolute path".to_string()));
+        }
+
+        let parsed = Url::parse(&self.broker)
+            .map_err(|e| TargetError::Configuration(format!("Invalid Pulsar broker URL: {e} (value: '{}')", self.broker)))?;
+        let tls_enabled = parsed.scheme() == "pulsar+ssl";
+        if !tls_enabled && (!self.tls_ca.is_empty() || self.tls_allow_insecure || !self.tls_hostname_verification) {
+            return Err(TargetError::Configuration(
+                "Pulsar TLS settings are only allowed with pulsar+ssl brokers".to_string(),
+            ));
+        }
+
+        Ok(())
+    }
+}
+
+pub fn validate_pulsar_broker(broker: &str) -> Result<Url, TargetError> {
+    let url = Url::parse(broker)
+        .map_err(|e| TargetError::Configuration(format!("Invalid Pulsar broker URL: {e} (value: '{broker}')")))?;
+
+    match url.scheme() {
+        "pulsar" | "pulsar+ssl" => {}
+        _ => {
+            return Err(TargetError::Configuration(
+                "Pulsar broker must use pulsar:// or pulsar+ssl://".to_string(),
+            ));
+        }
+    }
+
+    if !url.username().is_empty() || url.password().is_some() {
+        return Err(TargetError::Configuration(
+            "Pulsar broker URL must not embed username or password".to_string(),
+        ));
+    }
+
+    if url.host_str().is_none() {
+        return Err(TargetError::Configuration("Pulsar broker is missing host".to_string()));
+    }
+
+    Ok(url)
+}
+
+pub async fn connect_pulsar(args: &PulsarArgs) -> Result<Pulsar<TokioExecutor>, TargetError> {
+    args.validate()?;
+
+    let mut builder = Pulsar::builder(args.broker.clone(), TokioExecutor);
+
+    if !args.auth_token.is_empty() {
+        builder = builder.with_auth(Authentication {
+            name: "token".to_string(),
+            data: args.auth_token.clone().into_bytes(),
+        });
+    } else if !args.username.is_empty() {
+        builder =
+            builder.with_auth_provider(pulsar::authentication::basic::BasicAuthentication::new(&args.username, &args.password));
+    }
+
+    if !args.tls_ca.is_empty() {
+        builder = builder
+            .with_certificate_chain_file(&args.tls_ca)
+            .map_err(|e| TargetError::Configuration(format!("Failed to load Pulsar tls_ca: {e}")))?;
+    }
+
+    builder = builder
+        .with_allow_insecure_connection(args.tls_allow_insecure)
+        .with_tls_hostname_verification_enabled(args.tls_hostname_verification);
+
+    builder
+        .build()
+        .await
+        .map_err(|e| TargetError::Network(format!("Failed to connect to Pulsar broker: {e}")))
+}
+
+pub struct PulsarTarget<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    id: TargetID,
+    args: PulsarArgs,
+    client: Mutex<Option<Pulsar<TokioExecutor>>>,
+    producer: AsyncMutex<Option<Producer<TokioExecutor>>>,
+    store: Option<Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>>,
+    connected: AtomicBool,
+    delivery_counters: Arc<TargetDeliveryCounters>,
+    _phantom: std::marker::PhantomData<E>,
+}
+
+impl<E> PulsarTarget<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    pub fn clone_box(&self) -> Box<dyn Target<E> + Send + Sync> {
+        Box::new(PulsarTarget::<E> {
+            id: self.id.clone(),
+            args: self.args.clone(),
+            client: Mutex::new(self.client.lock().unwrap().clone()),
+            producer: AsyncMutex::new(None),
+            store: self.store.as_ref().map(|s| s.boxed_clone()),
+            connected: AtomicBool::new(self.connected.load(Ordering::SeqCst)),
+            delivery_counters: Arc::clone(&self.delivery_counters),
+            _phantom: std::marker::PhantomData,
+        })
+    }
+
+    #[instrument(skip(args), fields(target_id_as_string = %id))]
+    pub fn new(id: String, args: PulsarArgs) -> Result<Self, TargetError> {
+        args.validate()?;
+        let target_id = TargetID::new(id, ChannelTargetType::Pulsar.as_str().to_string());
+        let queue_store = if !args.queue_dir.is_empty() {
+            let base_path = PathBuf::from(&args.queue_dir);
+            let specific_queue_path = base_path.join(format!("rustfs-{}-{}", ChannelTargetType::Pulsar.as_str(), target_id.id));
+            let extension = match args.target_type {
+                TargetType::AuditLog => rustfs_config::audit::AUDIT_STORE_EXTENSION,
+                TargetType::NotifyEvent => rustfs_config::notify::NOTIFY_STORE_EXTENSION,
+            };
+            let store = QueueStore::<QueuedPayload>::new(specific_queue_path, args.queue_limit, extension);
+            if let Err(e) = store.open() {
+                error!(target_id = %target_id, error = %e, "Failed to open store for Pulsar target");
+                return Err(TargetError::Storage(format!("{e}")));
+            }
+            Some(Box::new(store) as Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync>)
+        } else {
+            None
+        };
+
+        Ok(Self {
+            id: target_id,
+            args,
+            client: Mutex::new(None),
+            producer: AsyncMutex::new(None),
+            store: queue_store,
+            connected: AtomicBool::new(false),
+            delivery_counters: Arc::new(TargetDeliveryCounters::default()),
+            _phantom: std::marker::PhantomData,
+        })
+    }
+
+    async fn get_or_connect_client(&self) -> Result<Pulsar<TokioExecutor>, TargetError> {
+        if let Some(client) = self.client.lock().unwrap().clone() {
+            return Ok(client);
+        }
+
+        let client = connect_pulsar(&self.args).await?;
+        self.connected.store(true, Ordering::SeqCst);
+        let mut guard = self.client.lock().unwrap();
+        let shared = guard.get_or_insert_with(|| client.clone()).clone();
+        Ok(shared)
+    }
+
+    async fn init_producer(&self) -> Result<(), TargetError> {
+        if self.producer.lock().await.is_some() {
+            return Ok(());
+        }
+
+        let client = self.get_or_connect_client().await?;
+        let producer = client
+            .producer()
+            .with_topic(self.args.topic.clone())
+            .with_name(self.id.id.clone())
+            .build()
+            .await
+            .map_err(|e| TargetError::Network(format!("Failed to create Pulsar producer: {e}")))?;
+
+        let mut guard = self.producer.lock().await;
+        if guard.is_none() {
+            *guard = Some(producer);
+        }
+        Ok(())
+    }
+
+    fn build_queued_payload(&self, event: &EntityTarget<E>) -> Result<QueuedPayload, TargetError> {
+        let object_name = crate::target::decode_object_name(&event.object_name)?;
+        let key = format!("{}/{}", event.bucket_name, object_name);
+        let log = TargetLog {
+            event_name: event.event_name,
+            key,
+            records: vec![event.clone()],
+        };
+        let body = serde_json::to_vec(&log).map_err(|e| TargetError::Serialization(format!("Failed to serialize event: {e}")))?;
+        let meta = QueuedPayloadMeta::new(
+            event.event_name,
+            event.bucket_name.clone(),
+            event.object_name.clone(),
+            "application/json",
+            body.len(),
+        );
+        Ok(QueuedPayload::new(meta, body))
+    }
+
+    async fn send_body(&self, body: Vec<u8>) -> Result<(), TargetError> {
+        self.init_producer().await?;
+        let mut guard = self.producer.lock().await;
+        let producer = guard
+            .as_mut()
+            .ok_or_else(|| TargetError::Configuration("Pulsar producer not initialized".to_string()))?;
+        let receipt = producer
+            .send_non_blocking(body)
+            .await
+            .map_err(|e| TargetError::Request(format!("Failed to send Pulsar message: {e}")))?;
+        receipt
+            .await
+            .map_err(|e| TargetError::Request(format!("Failed to receive Pulsar receipt: {e}")))?;
+        self.delivery_counters.record_success();
+        Ok(())
+    }
+}
+
+#[async_trait]
+impl<E> Target<E> for PulsarTarget<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    fn id(&self) -> TargetID {
+        self.id.clone()
+    }
+
+    async fn is_active(&self) -> Result<bool, TargetError> {
+        self.init_producer().await?;
+        let guard = self.producer.lock().await;
+        let producer = guard
+            .as_ref()
+            .ok_or_else(|| TargetError::Configuration("Pulsar producer not initialized".to_string()))?;
+        producer
+            .check_connection()
+            .await
+            .map_err(|e| TargetError::Network(format!("Pulsar health check failed: {e}")))?;
+        Ok(true)
+    }
+
+    async fn save(&self, event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
+        let queued = match self.build_queued_payload(&event) {
+            Ok(queued) => queued,
+            Err(err) => {
+                self.delivery_counters.record_final_failure();
+                return Err(err);
+            }
+        };
+
+        if let Some(store) = &self.store {
+            let encoded = match queued.encode() {
+                Ok(encoded) => encoded,
+                Err(err) => {
+                    self.delivery_counters.record_final_failure();
+                    return Err(TargetError::Storage(format!("Failed to encode queued payload: {err}")));
+                }
+            };
+            if let Err(e) = store.put_raw(&encoded) {
+                self.delivery_counters.record_final_failure();
+                return Err(TargetError::Storage(format!("Failed to save event to store: {e}")));
+            }
+            Ok(())
+        } else {
+            if let Err(err) = self.send_body(queued.body).await {
+                self.delivery_counters.record_final_failure();
+                return Err(err);
+            }
+            Ok(())
+        }
+    }
+
+    async fn send_raw_from_store(&self, _key: Key, body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
+        self.send_body(body).await
+    }
+
+    async fn close(&self) -> Result<(), TargetError> {
+        let mut producer = self.producer.lock().await;
+        if let Some(producer) = producer.as_mut() {
+            producer
+                .close()
+                .await
+                .map_err(|e| TargetError::Network(format!("Failed to close Pulsar producer: {e}")))?;
+        }
+        *producer = None;
+        self.client.lock().unwrap().take();
+        self.connected.store(false, Ordering::SeqCst);
+        info!(target_id = %self.id, "Pulsar target closed");
+        Ok(())
+    }
+
+    fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
+        self.store.as_deref()
+    }
+
+    fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
+        self.clone_box()
+    }
+
+    async fn init(&self) -> Result<(), TargetError> {
+        if !self.is_enabled() {
+            return Ok(());
+        }
+        self.init_producer().await
+    }
+
+    fn is_enabled(&self) -> bool {
+        self.args.enable
+    }
+
+    fn delivery_snapshot(&self) -> TargetDeliverySnapshot {
+        self.delivery_counters
+            .snapshot(self.store.as_deref().map_or(0, |store| store.len() as u64))
+    }
+
+    fn record_final_failure(&self) {
+        self.delivery_counters.record_final_failure();
+    }
+}