fix(ros2): address Copilot round-2 review on PR #9745

Follow-up to 7e5d2d2 that closes the remaining races and semantics
bugs surfaced by Copilot's second pass:

  * SubscriberImpl::GetMessage: clear _new_message under _message_mutex
    so on_data_available cannot drop a pending sample by racing the
    flag-clear against its own store(true). Same fix applied to the
    on_data_available side: the lock now spans both the _message write
    and the flag store.

  * SubscriberImpl::on_subscription_matched and
    PublisherImpl::on_publication_matched: derive _alive from
    info.current_count > 0, not info.total_count > 0. total_count is
    cumulative and never decrements, so the previous code latched
    IsAlive() to true forever after the first match.

  * ROS2::AddActorRosName: insert_or_assign so a re-registered actor
    actually picks up its new ros_name. The previous unordered_map
    ::insert silently kept the stale entry.

Author: JArmandoAnaya

LibCarla/source/carla/ros2/ROS2.cpp

@@ -132,7 +132,9 @@ void ROS2::SetTimestamp(double timestamp) {
 }
 
 void ROS2::AddActorRosName(void *actor, std::string ros_name) {
-  _actor_ros_name.insert({actor, ros_name});
+  // insert_or_assign so re-registering an actor with a new ros_name actually
+  // updates the entry; unordered_map::insert would silently keep the stale one.
+  _actor_ros_name.insert_or_assign(actor, std::move(ros_name));
 }
 
 void ROS2::AddActorParentRosName(void *actor, void* parent) {

LibCarla/source/carla/ros2/publishers/PublisherImpl.h

@@ -55,7 +55,7 @@ class PublisherImpl : public efd::DataWriterListener {
   void on_publication_matched(
       efd::DataWriter * /*writer*/,
       const efd::PublicationMatchedStatus &info) override {
-    _alive.store(info.total_count > 0, std::memory_order_release);
+    _alive.store(info.current_count > 0, std::memory_order_release);
   }
 
   ~PublisherImpl() override {

LibCarla/source/carla/ros2/subscribers/SubscriberImpl.h

@@ -56,14 +56,16 @@ class SubscriberImpl : public efd::DataReaderListener {
   void on_subscription_matched(
       efd::DataReader * /*reader*/,
       const efd::SubscriptionMatchedStatus &info) override {
-    _alive.store(info.total_count > 0, std::memory_order_release);
+    _alive.store(info.current_count > 0, std::memory_order_release);
   }
 
   void on_data_available(efd::DataReader *reader) override {
     // FastDDS invokes this on a DDS listener thread. Take into a stack-local
     // sample first so the lock only spans the copy into _message, and so a
     // dispose/unregister notification (RETCODE_OK with !info.valid_data) does
-    // not flip _new_message.
+    // not flip _new_message. The _new_message store is held inside the lock
+    // so a concurrent GetMessage cannot race-clear the flag after we publish
+    // a fresh sample.
     efd::SampleInfo info;
     msg_type sample{};
     erc rcode = reader->take_next_sample(&sample, &info);
@@ -74,10 +76,8 @@ class SubscriberImpl : public efd::DataReaderListener {
     if (!info.valid_data) {
       return;
     }
-    {
-      std::lock_guard<std::mutex> lock(_message_mutex);
-      _message = std::move(sample);
-    }
+    std::lock_guard<std::mutex> lock(_message_mutex);
+    _message = std::move(sample);
     _new_message.store(true, std::memory_order_release);
   }
 
@@ -141,11 +141,8 @@ class SubscriberImpl : public efd::DataReaderListener {
   [[nodiscard]] bool IsAlive() const noexcept { return _alive.load(std::memory_order_acquire); }
 
   msg_type GetMessage() {
-    msg_type copy;
-    {
-      std::lock_guard<std::mutex> lock(_message_mutex);
-      copy = _message;
-    }
+    std::lock_guard<std::mutex> lock(_message_mutex);
+    msg_type copy = _message;
     _new_message.store(false, std::memory_order_release);
     return copy;
   }