logging.rs

// Copyright (c) 2019 Sequence Planner
// SPDX-License-Identifier: Apache-2.0 AND MIT
// Adapted from https://github.com/sequenceplanner/r2r/blob/89cec03d07a1496a225751159cbc7bfb529d9dd1/r2r/src/utils.rs
// Further adapted from https://github.com/mvukov/rules_ros2/pull/371

use std::{
    collections::HashMap,
    ffi::CString,
    sync::{Mutex, OnceLock},
};

use crate::{rcl_bindings::*, ENTITY_LIFECYCLE_MUTEX};

mod logging_configuration;
pub(crate) use logging_configuration::*;

mod log_params;
pub use log_params::*;

mod logger;
pub use logger::*;

/// log a message to rosout
///
/// # Examples
///
/// ```
/// use rclrs::*;
/// use std::sync::Mutex;
/// use std::time::Duration;
/// use std::env;
///
/// let executor = rclrs::Context::default().create_basic_executor();
/// let node = executor.create_node("test_node").unwrap();
///
/// log!(node.debug(), "Simple debug message");
/// let some_variable = 43;
/// log!(node.debug(), "Formatted debug message: {some_variable}");
/// log!(node.fatal(), "Fatal message from {}", node.name());
/// log!(node.warn().once(), "Only log this the first time");
/// log!(
///     node
///     .error()
///     .skip_first()
///     .throttle(Duration::from_millis(1000)),
///     "Noisy error that we expect the first time"
/// );
///
/// let count = 0;
/// log!(
///     node
///     .info()
///     .throttle(Duration::from_millis(1000))
///     .only_if(count % 10 == 0),
///     "Manually constructed LogConditions",
/// );
/// ```
///
/// All of the above examples will also work with the severity-specific log macros,
/// but any severity that gets passed in will be overridden:
/// - [`log_debug`][crate::log_debug]
/// - [`log_info`][crate::log_info]
/// - [`log_warn`][crate::log_warn]
/// - [`log_error`][crate::log_error]
/// - [`log_fatal`][crate::log_fatal]
///
/// # Panics
///
/// It is theoretically possible for the call to panic if the Mutex used for the throttling is
/// poisoned although this should never happen.
#[macro_export]
macro_rules! log {
    // The variable args captured by the $(, $($args:tt)*)?)) code allows us to omit (or include)
    // formatting parameters in the simple message case, e.g. to write
    // ```
    // log_error!(<logger>, "Log with no params"); // OR
    // log_error!(<logger>, "Log with useful info {}", error_reason);
    ($to_log_params: expr, $($args:tt)*) => {{
        // Adding these use statements here due an issue like this one:
        // https://github.com/intellij-rust/intellij-rust/issues/9853
        // Note: that issue appears to be specific to jetbrains intellisense however,
        // observed same/similar behaviour with rust-analyzer/rustc
        use std::sync::{Once, OnceLock, Mutex};
        use std::time::{SystemTime, Instant};

        // We wrap the functional body of the macro inside of a closure which
        // we immediately trigger. This allows us to use `return` to exit the
        // macro early without causing the calling function to also try to
        // return early.
        (|| {
            let params = $crate::ToLogParams::to_log_params($to_log_params);

            if !params.get_user_filter() {
                // The user filter is telling us not to log this time, so exit
                // before doing anything else.
                return;
            }

            let mut first_time = false;
            static REMEMBER_FIRST_TIME: Once = Once::new();
            REMEMBER_FIRST_TIME.call_once(|| first_time = true);

            let logger_name = params.get_logger_name();
            let severity = params.get_severity();

            match params.get_occurence() {
                // Create the static variables here so we get a per-instance static
                $crate::LogOccurrence::Once => {
                    if first_time {
                        $crate::log_unconditional!(severity, logger_name, $($args)*);
                    }
                    // Since we've already logged once, we should never log again,
                    // so just exit right now.
                    return;
                }
                $crate::LogOccurrence::SkipFirst => {
                    if first_time {
                        // This is the first time that we're seeing this log, and we
                        // were told to skip the first one, so just exit right away.
                        return;
                    }
                }
                // Do nothing
                $crate::LogOccurrence::All => (),
            }

            // If we have a throttle duration then check if we're inside or outside
            // of that interval.
            let throttle = params.get_throttle();
            if throttle > std::time::Duration::ZERO {
                match params.get_throttle_clock() {
                    $crate::ThrottleClock::SteadyTime => {
                        static LAST_LOG_STEADY_TIME: OnceLock<Mutex<Instant>> = OnceLock::new();
                        let last_log_time = LAST_LOG_STEADY_TIME.get_or_init(|| {
                            Mutex::new(Instant::now())
                        });

                        if !first_time {
                            let now = Instant::now();
                            let mut previous = last_log_time.lock().unwrap();
                            if now >= *previous + throttle {
                                *previous = now;
                            } else {
                                // We are still inside the throttle interval, so just exit here.
                                return;
                            }
                        }
                    }
                    $crate::ThrottleClock::SystemTime => {
                        static LAST_LOG_SYSTEM_TIME: OnceLock<Mutex<SystemTime>> = OnceLock::new();
                        let last_log_time = LAST_LOG_SYSTEM_TIME.get_or_init(|| {
                            Mutex::new(SystemTime::now())
                        });

                        if !first_time {
                            let now = SystemTime::now();
                            let mut previous = last_log_time.lock().unwrap();
                            if now >= *previous + throttle {
                                *previous = now;
                            } else {
                                // We are still inside the throttle interval, so just exit here.
                                return;
                            }
                        }
                    }
                    $crate::ThrottleClock::Clock(clock) => {
                        static LAST_LOG_CLOCK_TIME: OnceLock<Mutex<$crate::Time>> = OnceLock::new();
                        let last_log_time = LAST_LOG_CLOCK_TIME.get_or_init(|| {
                            Mutex::new(clock.now())
                        });

                        if !first_time {
                            let now = clock.now();
                            let mut previous = last_log_time.lock().unwrap();

                            let new_interval = !now.compare_with(
                                &(previous.clone() + throttle),
                                |now, interval| now < interval,
                            )
                            .is_some_and(|eval| eval);

                            if new_interval {
                                *previous = now;
                            } else {
                                // We are still inside the throttle interval, so just exit here.
                                return;
                            }
                        }
                    }
                }
            }

            // All filters have been checked, so go ahead and publish the message
            $crate::log_unconditional!(severity, logger_name, $($args)*);
        })();
    }};
}

/// Debug log message. See [`log`] for usage.
#[macro_export]
macro_rules! log_debug {
    ($to_log_params: expr, $($args:tt)*) => {{
        let log_params = $crate::ToLogParams::to_log_params($to_log_params);
        $crate::log!($crate::ToLogParams::debug(log_params), $($args)*);
    }}
}

/// Info log message. See [`log`] for usage.
#[macro_export]
macro_rules! log_info {
    ($to_log_params: expr, $($args:tt)*) => {{
        let log_params = $crate::ToLogParams::to_log_params($to_log_params);
        $crate::log!($crate::ToLogParams::info(log_params), $($args)*);
    }}
}

/// Warning log message. See [`log`] for usage.
#[macro_export]
macro_rules! log_warn {
    ($to_log_params: expr, $($args:tt)*) => {{
        let log_params = $crate::ToLogParams::to_log_params($to_log_params);
        $crate::log!($crate::ToLogParams::warn(log_params), $($args)*);
    }}
}

/// Error log message. See [`log`] for usage.
#[macro_export]
macro_rules! log_error {
    ($to_log_params: expr, $($args:tt)*) => {{
        let log_params = $crate::ToLogParams::to_log_params($to_log_params);
        $crate::log!($crate::ToLogParams::error(log_params), $($args)*);
    }}
}

/// Fatal log message. See [`log`] for usage.
#[macro_export]
macro_rules! log_fatal {
    ($to_log_params: expr, $($args:tt)*) => {{
        let log_params = $crate::ToLogParams::to_log_params($to_log_params);
        $crate::log!($crate::ToLogParams::fatal(log_params), $($args)*);
    }}
}

/// A logging mechanism that does not have any conditions: It will definitely
/// publish the log. This is only meant for internal use, but needs to be exported
/// in order for [`log`] to work.
#[doc(hidden)]
#[macro_export]
macro_rules! log_unconditional {
    ($severity: expr, $logger_name: expr, $($args:tt)*) => {{
        use std::{ffi::CString, sync::OnceLock};

        // Only allocate a CString for the function name once per call to this macro.
        static FUNCTION_NAME: OnceLock<CString> = OnceLock::new();
        let function_name = FUNCTION_NAME.get_or_init(|| {
            // This call to function! is nested within two layers of closures,
            // so we need to strip away those suffixes or else users will be
            // misled. If we ever restructure these macros or if Rust changes
            // the way it names closures, this implementation detail may need to
            // change.
            let function_name = $crate::function!()
                .strip_suffix("::{{closure}}::{{closure}}")
                .unwrap();

            CString::new(function_name).unwrap_or(
                CString::new("<invalid name>").unwrap()
            )
        });

        // Only allocate a CString for the file name once per call to this macro.
        static FILE_NAME: OnceLock<CString> = OnceLock::new();
        let file_name = FILE_NAME.get_or_init(|| {
            CString::new(file!()).unwrap_or(
                CString::new("<invalid name>").unwrap()
            )
        });

        // We have to allocate a CString for the message every time because the
        // formatted data may have changed. We could consider having an alternative
        // macro for string literals that only allocates once, but it not obvious
        // how to guarantee that the user only passes in an unchanging string literal.
        match CString::new(std::fmt::format(format_args!($($args)*))) {
            Ok(message) => {
                // SAFETY: impl_log is actually completely safe to call, we just
                // mark it as unsafe to discourage downstream users from calling it
                unsafe { $crate::impl_log($severity, $logger_name, &message, &function_name, &file_name, line!()) };
            }
            Err(err) => {
                let message = CString::new(format!(
                    "Unable to format log message into a valid c-string. Error: {}", err
                )).unwrap();

                // SAFETY: impl_log is actually completely safe to call, we just
                // mark it as unsafe to discourage downstream users from calling it
                unsafe {
                    $crate::impl_log(
                        $crate::LogSeverity::Error,
                        &$crate::LoggerName::Unvalidated("logger"),
                        &message,
                        &function_name,
                        &file_name,
                        line!(),
                    );
                }
            }
        }
    }}
}

/// Calls the underlying rclutils logging function
/// Don't call this directly, use the logging macros instead, i.e. [`log`].
///
/// SAFETY: This function is not actually unsafe, but it is not meant to be part of the public
/// API, so we mark it as unsafe to discourage users from trying to use it. They should use
/// one of the of log! macros instead. We can't make it private because it needs to be used
/// by exported macros.
#[doc(hidden)]
pub unsafe fn impl_log(
    severity: LogSeverity,
    logger_name: &LoggerName,
    message: &CString,
    function: &CString,
    file: &CString,
    line: u32,
) {
    // We use a closure here because there are several different points in this
    // function where we may need to run this same logic.
    let send_log = |severity: LogSeverity, logger_name: &CString, message: &CString| {
        let location = rcutils_log_location_t {
            function_name: function.as_ptr(),
            file_name: file.as_ptr(),
            line_number: line as usize,
        };

        static FORMAT_STRING: OnceLock<CString> = OnceLock::new();
        let format_string = FORMAT_STRING.get_or_init(|| CString::new("%s").unwrap());

        let severity = severity.as_native() as i32;

        let _lifecycle = ENTITY_LIFECYCLE_MUTEX.lock().unwrap();

        #[cfg(test)]
        if log_handler::is_using_custom_handler()
            && unsafe { rcutils_logging_logger_is_enabled_for(logger_name.as_ptr(), severity) }
        {
            unsafe {
                // SAFETY: The global mutex is locked as _lifecycle and the pointers are valid for
                // the duration of this call.
                log_handler::dispatch_logging_output_handler(
                    &location,
                    severity,
                    logger_name.as_ptr(),
                    log_handler::null_timestamp(),
                    message.as_ptr(),
                );
            }
        }

        unsafe {
            // SAFETY: The global mutex is locked as _lifecycle
            rcutils_log(
                &location,
                severity,
                logger_name.as_ptr(),
                format_string.as_ptr(),
                message.as_ptr(),
            );
        }
    };

    match logger_name {
        LoggerName::Validated(c_name) => {
            // The logger name is pre-validated, so just go ahead and use it.
            send_log(severity, c_name, message);
        }
        LoggerName::Unvalidated(str_name) => {
            // The name was not validated before being passed in.
            //
            // We maintain a hashmap of previously validated loggers so
            // we don't need to reallocate the CString on every log instance.
            // This is done inside of the function impl_log instead of in a macro
            // so that this map is global for the entire application.
            static NAME_MAP: OnceLock<Mutex<HashMap<String, CString>>> = OnceLock::new();
            let name_map = NAME_MAP.get_or_init(Default::default);

            {
                // We need to keep name_map locked while we call send_log, but
                // we also need to make sure it gets unlocked right afterward
                // if the condition fails, otherwise this function would
                // deadlock on itself when handling the error case of the logger
                // name being invalid. So we keep name_map_guard in this extra
                // scope to isolate its lifespan.
                let name_map_guard = name_map.lock().unwrap();
                if let Some(c_name) = name_map_guard.get(*str_name) {
                    // The map name has been used before, so we just use the
                    // pre-existing CString
                    send_log(severity, c_name, message);

                    // We return right away because the remainder of this
                    // function just allocates and validates a new CString for
                    // the logger name.
                    return;
                }
            }

            // The unvalidated logger name has not been used before, so we need
            // to convert it and add it to the name_map now.
            let c_name = match CString::new(str_name.to_string()) {
                Ok(c_name) => c_name,
                Err(_) => {
                    static INVALID_MSG: OnceLock<CString> = OnceLock::new();
                    let invalid_msg = INVALID_MSG.get_or_init(|| {
                        CString::new(
                            "Failed to convert logger name into a c-string. \
                            Check for null terminators inside the string.",
                        )
                        .unwrap()
                    });
                    static INTERNAL_LOGGER_NAME: OnceLock<CString> = OnceLock::new();
                    let internal_logger_name =
                        INTERNAL_LOGGER_NAME.get_or_init(|| CString::new("logger").unwrap());
                    send_log(severity, internal_logger_name, invalid_msg);
                    return;
                }
            };

            send_log(severity, &c_name, message);
            name_map
                .lock()
                .unwrap()
                .insert(str_name.to_string(), c_name);
        }
    }
}

/// Used internally by logging macros to get the name of the function that called the
/// logging macro. This is not meant for public use, but we need to export it so the
/// other exported macros can use it. We should remove it if an official function! macro
/// is ever offered.
#[macro_export]
#[doc(hidden)]
macro_rules! function {
    () => {{
        fn f() {}
        fn type_name_of<T>(_: T) -> &'static str {
            std::any::type_name::<T>()
        }
        let name = type_name_of(f);
        name.strip_suffix("::f").unwrap()
    }};
}

#[cfg(test)]
mod tests {
    use crate::{log_handler::*, test_helpers::*, *};
    use std::{
        sync::{Arc, Mutex},
        time::Duration,
    };

    #[test]
    fn test_logging_macros() -> Result<(), RclrsError> {
        // This test ensures that strings which are being sent to the logger are
        // being sanitized correctly. Rust generally and our logging macro in
        // particular do not use C-style formatting strings, but rcutils expects
        // to receive C-style formatting strings alongside variadic arguments
        // that describe how to fill in the formatting.
        //
        // If we pass the final string into rcutils as the format with no
        // variadic arguments, then it may trigger a crash or undefined behavior
        // if the message happens to contain any % symbols. In particular %n
        // will trigger a crash when no variadic arguments are given because it
        // attempts to write to a buffer. If no buffer is given, a seg fault
        // happens.
        log!("please do not crash", "%n");

        let graph = construct_test_graph("test_logging_macros")?;

        let log_collection: Arc<Mutex<Vec<LogEntry<'static>>>> = Arc::new(Mutex::new(Vec::new()));
        let inner_log_collection = log_collection.clone();

        log_handler::set_logging_output_handler(move |log_entry: log_handler::LogEntry| {
            inner_log_collection
                .lock()
                .unwrap()
                .push(log_entry.into_owned());
        })
        .unwrap();

        let last_logger_name = || {
            log_collection
                .lock()
                .unwrap()
                .last()
                .unwrap()
                .logger_name
                .clone()
        };

        let last_message = || {
            log_collection
                .lock()
                .unwrap()
                .last()
                .unwrap()
                .message
                .clone()
        };

        let last_location = || {
            log_collection
                .lock()
                .unwrap()
                .last()
                .unwrap()
                .location
                .clone()
        };

        let last_severity = || log_collection.lock().unwrap().last().unwrap().severity;

        let count_message = |message: &str| {
            let mut count = 0;
            for log in log_collection.lock().unwrap().iter() {
                if log.message == message {
                    count += 1;
                }
            }
            count
        };

        let node = graph.node1;

        log!(&*node, "Logging with node dereference");
        assert_eq!(last_logger_name(), node.logger().name());
        assert_eq!(last_message(), "Logging with node dereference");
        assert_eq!(last_severity(), LogSeverity::Info);
        assert_eq!(
            last_location().function_name,
            "rclrs::logging::tests::test_logging_macros",
        );

        for _ in 0..10 {
            log!(node.once(), "Logging once");
        }
        assert_eq!(count_message("Logging once"), 1);
        assert_eq!(last_severity(), LogSeverity::Info);

        log!(node.logger(), "Logging with node logger");
        assert_eq!(last_message(), "Logging with node logger");
        assert_eq!(last_severity(), LogSeverity::Info);

        log!(node.debug(), "Debug from node");
        // The default severity level is Info so we should not see the last message
        assert_ne!(last_message(), "Debug from node");
        assert_ne!(last_severity(), LogSeverity::Debug);

        log!(node.info(), "Info from node");
        assert_eq!(last_message(), "Info from node");
        assert_eq!(last_severity(), LogSeverity::Info);

        log!(node.warn(), "Warn from node");
        assert_eq!(last_message(), "Warn from node");
        assert_eq!(last_severity(), LogSeverity::Warn);

        log!(node.error(), "Error from node");
        assert_eq!(last_message(), "Error from node");
        assert_eq!(last_severity(), LogSeverity::Error);

        log!(node.fatal(), "Fatal from node");
        assert_eq!(last_message(), "Fatal from node");
        assert_eq!(last_severity(), LogSeverity::Fatal);

        log_debug!(node.logger(), "log_debug macro");
        log_info!(node.logger(), "log_info macro");
        log_warn!(node.logger(), "log_warn macro");
        log_error!(node.logger(), "log_error macro");
        log_fatal!(node.logger(), "log_fatal macro");

        log!(node.only_if(false), "This should not be logged",);
        log!(node.only_if(true), "This should be logged",);

        for i in 0..3 {
            log!(node.warn().skip_first(), "Formatted warning #{}", i);
        }
        assert_eq!(count_message("Formatted warning #0"), 0);
        assert_eq!(count_message("Formatted warning #1"), 1);
        assert_eq!(count_message("Formatted warning #2"), 1);

        node.logger().set_level(LogSeverity::Debug).unwrap();
        log_debug!(node.logger(), "This debug message appears");
        assert_eq!(last_message(), "This debug message appears");
        assert_eq!(last_severity(), LogSeverity::Debug);

        node.logger().set_level(LogSeverity::Info).unwrap();
        log_debug!(node.logger(), "This debug message does not appear");
        assert_ne!(last_message(), "This debug message does not appear");

        log!("custom logger name", "message for custom logger");
        assert_eq!(last_logger_name(), "custom logger name");
        assert_eq!(last_message(), "message for custom logger");

        for _ in 0..3 {
            log!(
                "custom logger name once".once(),
                "one-time message for custom logger",
            );
        }
        assert_eq!(last_logger_name(), "custom logger name once");
        assert_eq!(last_severity(), LogSeverity::Info);
        assert_eq!(count_message("one-time message for custom logger"), 1);

        for _ in 0..3 {
            log!(
                "custom logger name skip".error().skip_first(),
                "error for custom logger",
            );
        }
        assert_eq!(last_logger_name(), "custom logger name skip");
        assert_eq!(last_severity(), LogSeverity::Error);
        assert_eq!(count_message("error for custom logger"), 2);

        // Test whether throttling works correctly with a ROS clock
        let (clock, source) = Clock::with_source();
        source.set_ros_time_override(0);

        for i in 0..15 {
            log!(
                "logger"
                    .throttle(Duration::from_nanos(10))
                    .throttle_clock(ThrottleClock::Clock(&clock)),
                "custom clock throttled message",
            );
            source.set_ros_time_override(i);
        }

        // The throttle interval is 10ns and the loop shifted the time from 0ns
        // to 14ns, triggering the log macro once per nanosecond. That means we
        // should see two messages in the log.
        assert_eq!(count_message("custom clock throttled message"), 2);

        reset_logging_output_handler();
        Ok(())
    }

    #[test]
    fn test_function_macro() {
        assert_eq!(function!(), "rclrs::logging::tests::test_function_macro");
    }

    #[test]
    fn test_rosout_publishing_default() -> Result<(), RclrsError> {
        use crate::rcl_bindings::rcl_logging_rosout_enabled;
        use ros_env::rcl_interfaces::msg::rmw::Log;
        use std::sync::{Arc, Mutex};

        let namespace = format!("/test_rosout_publishing_default_{}", line!());
        let mut executor = Context::default().create_basic_executor();
        let node = executor
            .create_node("rosout_default_node".namespace(&namespace))
            .unwrap();

        // Check if rosout is enabled at the rcl level
        // SAFETY: This is a simple query function with no preconditions
        let rosout_enabled = unsafe { rcl_logging_rosout_enabled() };
        assert!(
            rosout_enabled,
            "rcl_logging_rosout should be enabled for this test"
        );

        // Verify rosout publisher exists immediately after node creation
        // (when enable_rosout is true, which is the default)
        let topics = node
            .get_publisher_names_and_types_by_node("rosout_default_node", &namespace)
            .unwrap();
        assert!(
            topics.contains_key("/rosout"),
            "Node should have /rosout publisher immediately when enable_rosout is true"
        );

        let received_logs: Arc<Mutex<Vec<Log>>> = Arc::new(Mutex::new(Vec::new()));
        let inner_received = Arc::clone(&received_logs);

        // Subscribe to /rosout topic with QoS matching rosout publisher
        // rosout uses: reliable, transient_local, keep_last(1000)
        // Filter by logger name to avoid interference from parallel tests
        let _subscription = node.create_subscription::<Log, _>(
            "/rosout".reliable().transient_local().keep_all(),
            move |msg: Log| {
                // Only count messages from our node to avoid interference from parallel tests
                if msg.name.to_string().contains("rosout_default_node") {
                    inner_received.lock().unwrap().push(msg);
                }
            },
        )?;

        // Wait for /rosout subscription to be discovered by the publisher
        spin_until_condition(
            &mut executor,
            || node.count_subscriptions("/rosout").unwrap_or(0) > 0,
            Duration::from_millis(500),
        );

        // Log messages at all severity levels (except debug) with unique identifiers
        let test_id = line!();
        let info_msg = format!("Info rosout default test {}", test_id);
        let warn_msg = format!("Warn rosout default test {}", test_id);
        let error_msg = format!("Error rosout default test {}", test_id);
        let fatal_msg = format!("Fatal rosout default test {}", test_id);

        log!(node.info(), "{}", info_msg);
        log!(node.warn(), "{}", warn_msg);
        log!(node.error(), "{}", error_msg);
        log!(node.fatal(), "{}", fatal_msg);

        // Spin until all 4 messages are received (INFO, WARN, ERROR, FATAL)
        spin_until_condition(
            &mut executor,
            || received_logs.lock().unwrap().len() >= 4,
            Duration::from_secs(5),
        );

        // Verify all messages were received with correct severity levels
        let logs = received_logs.lock().unwrap();

        let find_log =
            |msg: &str| -> Option<&Log> { logs.iter().find(|l| l.msg.to_string() == msg) };

        // Verify INFO message
        let info_log = find_log(&info_msg);
        assert!(
            info_log.is_some(),
            "Info message not found on /rosout. Received {} messages: {:?}",
            logs.len(),
            logs.iter().map(|l| l.msg.to_string()).collect::<Vec<_>>()
        );
        assert_eq!(info_log.unwrap().level, Log::INFO);

        // Verify WARN message
        let warn_log = find_log(&warn_msg);
        assert!(warn_log.is_some(), "Warn message not found on /rosout");
        assert_eq!(warn_log.unwrap().level, Log::WARN);

        // Verify ERROR message
        let error_log = find_log(&error_msg);
        assert!(error_log.is_some(), "Error message not found on /rosout");
        assert_eq!(error_log.unwrap().level, Log::ERROR);

        // Verify FATAL message
        let fatal_log = find_log(&fatal_msg);
        assert!(fatal_log.is_some(), "Fatal message not found on /rosout");
        assert_eq!(fatal_log.unwrap().level, Log::FATAL);

        // Verify logger name is correct for all messages
        for log in logs.iter() {
            assert!(
                log.name.to_string().contains("rosout_default_node"),
                "Logger name '{}' should contain 'rosout_default_node'",
                log.name
            );
        }

        // Verify rosout publisher still exists after logging
        let topics = node
            .get_publisher_names_and_types_by_node("rosout_default_node", &namespace)
            .unwrap();
        assert!(
            topics.contains_key("/rosout"),
            "Node should still have /rosout publisher after logging"
        );

        Ok(())
    }

    #[test]
    fn test_rosout_disabled() -> Result<(), RclrsError> {
        use ros_env::rcl_interfaces::msg::rmw::Log;
        use std::sync::{
            atomic::{AtomicBool, Ordering},
            Arc,
        };

        let namespace = format!("/test_rosout_disabled_{}", line!());
        let mut executor = Context::default().create_basic_executor();

        // Create node with rosout DISABLED
        let node_disabled = executor
            .create_node(
                "rosout_disabled_node"
                    .namespace(&namespace)
                    .enable_rosout(false),
            )
            .unwrap();

        // Create another node with rosout ENABLED to subscribe
        let node_subscriber = executor
            .create_node("rosout_subscriber_node".namespace(&namespace))
            .unwrap();

        let test_id = line!();
        let disabled_msg = format!("This should NOT appear {}", test_id);
        let disabled_msg_clone = disabled_msg.clone();

        let received_from_disabled = Arc::new(AtomicBool::new(false));
        let inner_received = Arc::clone(&received_from_disabled);

        // Subscribe to /rosout topic with QoS matching rosout publisher
        let _subscription = node_subscriber.create_subscription::<Log, _>(
            "/rosout".reliable().transient_local().keep_all(),
            move |msg: Log| {
                if msg.msg.to_string() == disabled_msg_clone {
                    inner_received.store(true, Ordering::Release);
                }
            },
        )?;

        // Log a message from the subscriber node to ensure rosout publisher exists
        log!(node_subscriber.info(), "Subscriber node rosout init");

        // Wait for discovery between subscriber's publisher and our subscription
        spin_until_condition(
            &mut executor,
            || node_subscriber.count_subscriptions("/rosout").unwrap_or(0) > 0,
            Duration::from_millis(500),
        );

        // Log a message from the node with rosout disabled
        log!(node_disabled.info(), "{}", disabled_msg);

        // Spin for a reasonable time to ensure message would have been received if published
        spin_until_condition(&mut executor, || false, Duration::from_millis(500));

        // Verify the message was NOT received
        assert!(
            !received_from_disabled.load(Ordering::Acquire),
            "Message from node with rosout disabled should NOT appear on /rosout"
        );

        // Verify the disabled node does not have a /rosout publisher
        let topics = node_disabled
            .get_publisher_names_and_types_by_node("rosout_disabled_node", &namespace)
            .unwrap();
        assert!(
            !topics.contains_key("/rosout"),
            "Node with rosout disabled should not have /rosout publisher"
        );

        Ok(())
    }
}