feat(LibCarla/ros2): add CycloneDDS CDR middleware, fix FastDDS large-message publish, add ROS2 smoke test

Replace CycloneDDS publisher/subscriber stubs with real implementations that
transport raw CDR bytes via dds_writecdr() / dds_takecdr(), bypassing the
CycloneDDS type system entirely. No IDL-generated files, no cyclonedds-cxx.

Add CycloneDDSSertype.h, a custom ddsi_sertype + ddsi_serdata for raw CDR
passthrough. CDR bytes (including the 4-byte DDS encapsulation header) are
stored inline after the ddsi_serdata header in a single malloc block.
Full ddsi_serdata_ops (15 entries) and ddsi_sertype_ops (14 entries) tables
are implemented against CycloneDDS 0.10.5. Public API: carla_cdr_create_topic()
registers the sertype with a participant, carla_cdr_wrap() wraps pre-serialized
CDR in a ddsi_serdata* for dds_writecdr(), carla_cdr_data()/carla_cdr_size()
expose received bytes for deserialize_from_cdr().

CycloneDDSPublisherMiddleware<T>: Init() creates participant/topic/writer,
Publish() calls serialize_to_cdr(*msg) then dds_writecdr().
CycloneDDSSubscriberMiddleware<T>: Init() creates participant/topic/reader
with data_available listener, listener calls dds_takecdr() then
deserialize_from_cdr() into the caller-supplied message pointer.

Both FastDDS and CycloneDDS are always built when --ros2 is specified.
FastDDS is a hard dependency (libfastcdr.a is used by CdrSerialization.h
for both backends). Setup.sh downloads and builds both libraries
unconditionally, LibCarla/cmake/CMakeLists.txt unconditionally adds both
fast_dds and cyclone_dds subdirs. New LibCarla/cmake/cyclone_dds/CMakeLists.txt
builds libcarla_cyclonedds.a linking libddsc.a + libfastcdr.a.

Add -DBUILD_DDSPERF=OFF to CycloneDDS cmake flags in Setup.sh to prevent
ddsperf build failure when -DBUILD_IDLC=OFF (ddsperf depends on idlc).

Fix GenericCdrPubSubType::getSerializedSizeProvider() to return the actual
per-instance serialized size instead of a fixed CdrTopicInfo::max_serialized_size()
value. The fixed value (648 bytes for Image, 27597 for PointCloud2) was far
smaller than real camera frames (~1-8 MB) or LiDAR scans (~1-20 MB), causing
FastDDS DataWriter::write() to fail with RETCODE_ERROR (code 1) on every tick.
Add cdr_serialized_size() to CdrSerialization.h to compute the actual size, and
change serialize() to use an auto-growing FastBuffer to eliminate the fixed-size
buffer constraint entirely.

Fix MultiStreamState::IsEnabledForROS() to check the _enable_for_ros set
directly instead of iterating sessions that all return false from the default
virtual implementation. The Python is_enabled_for_ros() API was always returning
false even after enable_for_ros() because of this.

Fix CarlaCameraPublisher.cpp and CarlaRadarPublisher.cpp: add explicit
#include <cmath>, the CycloneDDS build exposed a pre-existing missing include
(FastDDS headers were providing std::tan/sin/cos transitively).

Add PythonAPI/test/smoke/test_ros2.py: two smoke tests for the native ROS2
publish path. test_ros2_api verifies the enable/disable/is_enabled lifecycle.
test_ros2_sensor_publish spawns an 800x600 RGB camera and LiDAR on a vehicle,
enables ROS2, ticks 20 synchronous frames, and asserts Python callbacks fire,
exercising the full GenericCdrPubSubType::serialize() -> DataWriter path with
realistic large payloads.

Tests: 120/120 server + 56/56 client. 2/2 smoke tests (smoke.test_ros2).

Author: JesusAnaya

LibCarla/cmake/CMakeLists.txt

@@ -26,7 +26,8 @@ elseif (CMAKE_BUILD_TYPE STREQUAL "Server")
 elseif (CMAKE_BUILD_TYPE STREQUAL "Pytorch")
 add_subdirectory("pytorch")
 elseif (CMAKE_BUILD_TYPE STREQUAL "ros2")
-add_subdirectory("fast_dds")
+  add_subdirectory("fast_dds")
+  add_subdirectory("cyclone_dds")
 else ()
   message(FATAL_ERROR "Unknown build type '${CMAKE_BUILD_TYPE}'")
 endif ()

LibCarla/cmake/cyclone_dds/CMakeLists.txt

@@ -0,0 +1,76 @@
+cmake_minimum_required(VERSION 3.5.1)
+project(libcarla_cyclonedds)
+
+# Install headers.
+
+file(GLOB libcarla_carla_cyclonedds_headers
+  "${libcarla_source_path}/carla/ros2/*.h"
+  "${libcarla_source_path}/carla/ros2/publishers/*.h"
+  "${libcarla_source_path}/carla/ros2/subscribers/*.h"
+  "${libcarla_source_path}/carla/ros2/listeners/*.h"
+  "${libcarla_source_path}/carla/ros2/types/*.h"
+  "${libcarla_source_path}/carla/ros2/types/msg/*.h"
+  "${libcarla_source_path}/carla/ros2/dds/*.h"
+  "${libcarla_source_path}/carla/ros2/dds/cyclonedds/*.h"
+  )
+install(FILES ${libcarla_carla_cyclonedds_headers} DESTINATION include/carla/ros2)
+
+file(GLOB cyclonedds_dependencies "${CYCLONEDDS_LIB_PATH}/*.a")
+install(FILES ${cyclonedds_dependencies} DESTINATION lib)
+
+file(GLOB libcarla_cyclonedds_sources
+  "${libcarla_source_path}/carla/ros2/*.cpp"
+  "${libcarla_source_path}/carla/ros2/publishers/*.cpp"
+  "${libcarla_source_path}/carla/ros2/subscribers/*.cpp"
+  "${libcarla_source_path}/carla/ros2/listeners/*.cpp"
+  "${libcarla_source_path}/carla/ros2/types/*.cpp")
+
+# ==============================================================================
+# Create targets for debug and release in the same build type.
+# ==============================================================================
+
+if (LIBCARLA_BUILD_RELEASE)
+  add_library(carla_cyclonedds STATIC ${libcarla_cyclonedds_sources})
+
+  target_compile_definitions(carla_cyclonedds PRIVATE CARLA_ROS2_DDS_CYCLONEDDS)
+  target_compile_options(carla_cyclonedds PRIVATE -fexceptions)
+
+  target_include_directories(carla_cyclonedds SYSTEM PRIVATE
+      "${BOOST_INCLUDE_PATH}"
+      "${RPCLIB_INCLUDE_PATH}")
+
+  # CycloneDDS C headers (dds/dds.h, dds/ddsi/*.h)
+  target_include_directories(carla_cyclonedds PRIVATE "${CYCLONEDDS_INCLUDE_PATH}")
+  # Fast-CDR headers (fastcdr/Cdr.h) — used by CdrSerialization.h
+  target_include_directories(carla_cyclonedds PRIVATE "${FASTDDS_INCLUDE_PATH}")
+
+  target_link_libraries(carla_cyclonedds
+      "${CYCLONEDDS_LIB_PATH}/libddsc.a"
+      "${FASTDDS_LIB_PATH}/libfastcdr.a")
+  install(TARGETS carla_cyclonedds DESTINATION lib)
+  set_target_properties(carla_cyclonedds PROPERTIES COMPILE_FLAGS "${CMAKE_CXX_FLAGS_RELEASE}")
+
+endif()
+
+if (LIBCARLA_BUILD_DEBUG)
+
+  add_library(carla_cyclonedds_debug STATIC ${libcarla_cyclonedds_sources})
+
+  target_compile_definitions(carla_cyclonedds_debug PRIVATE CARLA_ROS2_DDS_CYCLONEDDS)
+  target_compile_options(carla_cyclonedds_debug PRIVATE -fexceptions)
+
+  target_include_directories(carla_cyclonedds_debug SYSTEM PRIVATE
+      "${BOOST_INCLUDE_PATH}"
+      "${RPCLIB_INCLUDE_PATH}")
+
+  target_include_directories(carla_cyclonedds_debug PRIVATE "${CYCLONEDDS_INCLUDE_PATH}")
+  target_include_directories(carla_cyclonedds_debug PRIVATE "${FASTDDS_INCLUDE_PATH}")
+
+  target_link_libraries(carla_cyclonedds_debug
+      "${CYCLONEDDS_LIB_PATH}/libddsc.a"
+      "${FASTDDS_LIB_PATH}/libfastcdr.a")
+  install(TARGETS carla_cyclonedds_debug DESTINATION lib)
+  set_target_properties(carla_cyclonedds_debug PROPERTIES COMPILE_FLAGS "${CMAKE_CXX_FLAGS_DEBUG}")
+  target_compile_definitions(carla_cyclonedds_debug PUBLIC -DBOOST_ASIO_ENABLE_BUFFER_DEBUGGING)
+
+endif()

LibCarla/source/carla/ros2/dds/cyclonedds/CycloneDDSPublisherMiddleware.h

@@ -5,41 +5,131 @@
 #pragma once
 
 #include "carla/ros2/dds/IDDSPublisherMiddleware.h"
+#include "carla/ros2/dds/cyclonedds/CycloneDDSSertype.h"
+#include "carla/ros2/types/CdrSerialization.h"
+#include "carla/ros2/types/CdrTopicInfo.h"
 #include "carla/Logging.h"
 
+#ifndef CARLA_ROS2_DDS_TESTING
+#ifdef CARLA_ROS2_DDS_CYCLONEDDS
+
+#include <dds/dds.h>
+#include <dds/ddsi/ddsi_serdata.h>
+#include <dds/ddsi/ddsi_sertype.h>
+
 namespace carla {
 namespace ros2 {
 
-/// CycloneDDS implementation of IDDSPublisherMiddleware (stub).
+/// CycloneDDS implementation of IDDSPublisherMiddleware.
 ///
-/// This is a placeholder that compiles and satisfies the factory interface
-/// but does not contain a real CycloneDDS implementation. All operations
-/// log an error and return failure. The real implementation will replace
-/// this file once CycloneDDS types and the C API are available.
+/// Serializes msg::* POD structs to CDR via CdrSerialization.h and
+/// publishes them as raw CDR bytes using dds_writecdr().  No IDL-generated
+/// type files or CycloneDDS C++ bindings are required.
 ///
 /// Parameterized on a traits type T that provides:
-///   T::msg_type  — middleware-neutral POD message struct
+///   T::msg_type  — the message type (a carla::ros2::msg::* POD struct)
 template<typename T>
 class CycloneDDSPublisherMiddleware : public IDDSPublisherMiddleware {
  public:
+  using msg_type = typename T::msg_type;
+
+  ~CycloneDDSPublisherMiddleware() override {
+    if (_participant != DDS_ENTITY_NIL) {
+      dds_delete(_participant);  // cascades to writer, topic, sertype
+    }
+  }
+
   bool Init(const std::string& topic_name) override {
-    log_error("CycloneDDSPublisherMiddleware: stub — not yet implemented "
-              "(topic '", topic_name, "')");
-    return false;
+    _participant = dds_create_participant(DDS_DOMAIN_DEFAULT, nullptr, nullptr);
+    if (_participant < 0) {
+      log_error("CycloneDDSPublisherMiddleware: Failed to create participant "
+                "(topic '", topic_name, "', code:", _participant, ")");
+      return false;
+    }
+
+    const char* type_name = CdrTopicInfo<msg_type>::type_name();
+    _topic = carla_cdr_create_topic(
+        _participant, topic_name.c_str(), type_name, &_sertype);
+    if (_topic < 0) {
+      log_error("CycloneDDSPublisherMiddleware: Failed to create topic '",
+                topic_name, "' (code:", _topic, ")");
+      return false;
+    }
+
+    dds_qos_t* qos = dds_create_qos();
+    dds_qset_reliability(qos, DDS_RELIABILITY_RELIABLE,
+                         DDS_SECS(1));
+    dds_qset_history(qos, DDS_HISTORY_KEEP_LAST, 1);
+    dds_listener_t* listener = dds_create_listener(this);
+    dds_lset_publication_matched(listener, carla_on_publication_matched);
+
+    _writer = dds_create_writer(_participant, _topic, qos, listener);
+    dds_delete_listener(listener);
+    dds_delete_qos(qos);
+    if (_writer < 0) {
+      log_error("CycloneDDSPublisherMiddleware: Failed to create writer "
+                "(topic '", topic_name, "', code:", _writer, ")");
+      return false;
+    }
+
+    _topic_name = topic_name;
+    return true;
   }
 
-  bool Publish(void* /*message_data*/) override {
-    return false;
+  bool Publish(void* message_data) override {
+    const msg_type* msg = static_cast<const msg_type*>(message_data);
+    std::vector<uint8_t> cdr = serialize_to_cdr(*msg);
+
+    struct ddsi_serdata* sd = carla_cdr_wrap(
+        _sertype,
+        cdr.data(),
+        static_cast<uint32_t>(cdr.size()));
+    if (!sd) {
+      log_error("CycloneDDSPublisherMiddleware::Publish (", _topic_name,
+                "): allocation failed");
+      return false;
+    }
+
+    dds_return_t rc = dds_writecdr(_writer, sd);
+    ddsi_serdata_unref(sd);
+    if (rc != DDS_RETCODE_OK) {
+      log_error("CycloneDDSPublisherMiddleware::Publish (", _topic_name,
+                "): dds_writecdr failed (code:", rc, ")");
+      return false;
+    }
+    return true;
   }
 
   bool IsAlive() const override {
-    return false;
+    return _alive;
   }
 
   std::string GetTopicName() const override {
-    return {};
+    return _topic_name;
+  }
+
+ private:
+  static void carla_on_publication_matched(
+      dds_entity_t /*writer*/,
+      const dds_publication_matched_status_t status,
+      void* arg)
+  {
+    CycloneDDSPublisherMiddleware* self =
+        static_cast<CycloneDDSPublisherMiddleware*>(arg);
+    self->_alive = (status.current_count > 0);
   }
+
+  dds_entity_t          _participant { DDS_ENTITY_NIL };
+  dds_entity_t          _topic       { DDS_ENTITY_NIL };
+  dds_entity_t          _writer      { DDS_ENTITY_NIL };
+  struct ddsi_sertype*  _sertype     { nullptr };
+
+  std::string _topic_name;
+  bool        _alive { false };
 };
 
 } // namespace ros2
 } // namespace carla
+
+#endif // CARLA_ROS2_DDS_CYCLONEDDS
+#endif // !CARLA_ROS2_DDS_TESTING