refactor(rtps)!: consolidate to a single rtps component + thorough docs (mermaid)

.github/workflows/build.yml

@@ -213,8 +213,6 @@ jobs:
           target: esp32s3
         - path: 'components/rtps/example'
           target: esp32
-        - path: 'components/rtps_embedded/example'
-          target: esp32
         - path: 'components/rtsp/example'
           target: esp32
         - path: 'components/runqueue/example'

.github/workflows/rtps_interop.yml

@@ -7,12 +7,11 @@ permissions:
 on:
   pull_request:
     paths:
-      - "components/rtps_embedded/**"
       - "components/rtps/**"
       - "components/socket/**"
       - "components/cdr/**"
       - "lib/espp.cmake"
-      - "pc/tests/rtps_embedded_*"
+      - "pc/tests/rtps_*"
       - ".github/workflows/rtps_interop.yml"
   workflow_dispatch:
 
@@ -36,5 +35,5 @@ jobs:
           submodules: "recursive"
       - name: Run interop matrix
         run: |
-          cd components/rtps_embedded/interop
+          cd components/rtps/interop
           ./run.sh

components/esp32-p4-function-ev-board/example/main/esp32_p4_function_ev_board_example.cpp

@@ -22,7 +22,7 @@
 
 #include "cdr.hpp"
 #include "ping.hpp"
-#include "rtps.hpp"
+#include "rtps_participant.hpp"
 
 #include "esp32-p4-function-ev-board.hpp"
 
@@ -31,11 +31,6 @@
 using namespace std::chrono_literals;
 using Board = espp::Esp32P4FunctionEvBoard;
 
-// Address of the most recently discovered RTPS peer (filled by the participant's
-// on_participant_discovered callback, read by the ping self-test).
-static std::mutex g_peer_mutex;
-static std::string g_peer_addr;
-
 static std::vector<uint8_t> audio_bytes;
 
 static bool load_audio(size_t &out_size, size_t &out_sample_rate);
@@ -363,18 +358,16 @@ extern "C" void app_main(void) {
     logger.info("BOOT button not initialized (shared with Ethernet RMII TXD1 pin)");
   }
 
-  // Connectivity self-test: once we have an IP (and a moment for RTPS discovery),
-  // ping the gateway and the discovered peer once, then stop. This makes it easy
-  // to tell board-vs-network problems apart (e.g. gateway reachable but peer not
-  // => client isolation / L2 reachability problem, not the board).
+  // Connectivity self-test: once we have an IP, ping the gateway once, then stop.
+  // This makes it easy to tell board-vs-network problems apart.
   espp::Task ping_task(
       espp::Task::Config{.callback = [&logger](std::mutex &m, std::condition_variable &cv) -> bool {
                            if (!have_ip) {
                              std::unique_lock<std::mutex> lk(m);
                              cv.wait_for(lk, 250ms);
                              return false; // keep waiting for an IP
                            }
-                           // give RTPS discovery a few seconds to find the peer
+                           // let the link settle for a few seconds before pinging
                            {
                              std::unique_lock<std::mutex> lk(m);
                              cv.wait_for(lk, 4s);
@@ -387,16 +380,6 @@ extern "C" void app_main(void) {
                            char gw[16] = {0};
                            esp_ip4addr_ntoa(&ip_info.gw, gw, sizeof(gw));
                            ping_target(logger, "gateway", gw);
-                           std::string peer;
-                           {
-                             std::lock_guard<std::mutex> lk(g_peer_mutex);
-                             peer = g_peer_addr;
-                           }
-                           if (!peer.empty()) {
-                             ping_target(logger, "peer", peer);
-                           } else {
-                             logger.warn("Ping self-test: no RTPS peer discovered yet to ping");
-                           }
                            logger.info("=== Connectivity self-test done ===");
                            return true; // one-shot
                          },
@@ -412,7 +395,6 @@ extern "C" void app_main(void) {
   const std::string topic = "espp/test/counter";
   const std::string rtps_type = "std_msgs::msg::dds_::UInt32_";
   uint32_t value = 0;
-  bool published = false;
   static constexpr auto loop_tick = 20ms; // RTPS loop tick
   static constexpr int64_t publish_period_us = 50'000'000;
   int64_t last_publish_us = 0;
@@ -426,29 +408,19 @@ extern "C" void app_main(void) {
       std::string address = ip_str;
       logger.info("Got IP {}, starting RTPS participant", address);
       participant = std::make_shared<espp::RtpsParticipant>(espp::RtpsParticipant::Config{
-          .node_name = "espp_publisher",
-          .participant_id = 10,
-          .advertised_address = address,
-          .announce_period = 500ms,
-          .on_participant_discovered =
-              [&logger](const auto &p) {
-                {
-                  std::lock_guard<std::mutex> lk(g_peer_mutex);
-                  if (g_peer_addr.empty()) {
-                    g_peer_addr = p.address;
-                  }
-                }
-                logger.info("discovered participant at {}", p.address);
-              },
-          .on_endpoint_discovered =
-              [&logger](const auto &endpoint) {
-                logger.info("discovered {} '{}'", endpoint.is_reader ? "reader" : "writer",
-                            endpoint.topic_name);
+          .interface_address = address,
+          // A remote reader (e.g. a ROS 2 subscriber) matched our writer: we now
+          // have a peer to publish to. The facade signals matches through this
+          // callback (it runs on an engine worker thread).
+          .on_publisher_matched =
+              [&logger]() {
+                rtps_has_peers = true;
+                logger.info("RTPS writer matched a remote reader");
               },
           .log_level = espp::Logger::Verbosity::DEBUG,
       });
       participant->add_writer({
-          .topic_name = topic,
+          .topic = topic,
           .type_name = rtps_type,
       });
       if (!participant->start()) {
@@ -460,18 +432,18 @@ extern "C" void app_main(void) {
     } else if (did_have_ip && !have_ip) {
       logger.warn("Lost IP, stopping RTPS participant");
       participant.reset();
+      rtps_has_peers = false;
       did_have_ip = false;
     }
     rtps_running = (participant != nullptr);
 
     // Publish the next counter value at 2 Hz (independent of the status refresh).
     // Only publish if there is a discovered peer (otherwise the publish() call will return false).
     bool publish_period_elapsed = (now_us - last_publish_us) >= publish_period_us;
-    bool can_publish =
-        participant && !participant->discovered_participants().empty() && publish_period_elapsed;
+    bool can_publish = participant && rtps_has_peers.load() && publish_period_elapsed;
     if (can_publish) {
       last_publish_us = now_us;
-      published = participant->publish(topic, serialize_uint32(value));
+      bool published = participant->publish(topic, serialize_uint32(value));
       if (published) {
         logger.info("published {}", value);
         ++value;
@@ -480,9 +452,9 @@ extern "C" void app_main(void) {
       }
     }
 
-    // Publish the RTPS counter/peer state for the status task to render.
+    // Publish the RTPS counter value for the status task to render (peer state
+    // is driven by on_publisher_matched above).
     rtps_value = value;
-    rtps_has_peers = published;
 
     // Stream any active playback to the speaker in chunks, advancing by
     // however much the stream buffer accepted

components/rtps/CMakeLists.txt

@@ -1,4 +1,65 @@
 idf_component_register(
-  INCLUDE_DIRS "include"
-  SRC_DIRS "src"
-  REQUIRES base_component cdr task socket)
+  SRCS
+    "src/rtps_participant.cpp"
+    "src/communication/EsppTransport.cpp"
+    "src/discovery/ParticipantProxyData.cpp"
+    "src/discovery/SEDPAgent.cpp"
+    "src/discovery/SPDPAgent.cpp"
+    "src/discovery/TopicData.cpp"
+    "src/entities/Domain.cpp"
+    "src/entities/Participant.cpp"
+    "src/entities/Reader.cpp"
+    "src/entities/StatefulReader.cpp"
+    "src/entities/StatefulWriter.cpp"
+    "src/entities/StatelessReader.cpp"
+    "src/entities/StatelessWriter.cpp"
+    "src/entities/Writer.cpp"
+    "src/messages/MessageReceiver.cpp"
+    "src/messages/MessageTypes.cpp"
+    "src/utils/Diagnostics.cpp"
+  INCLUDE_DIRS
+    "include"
+  REQUIRES
+    base_component cdr task thread_pool socket
+)
+
+# Select the RTPS static-limits profile from Kconfig (see Kconfig in this
+# component). The "embedded" profile is the default and is byte-identical to the
+# historical behavior: it defines no RTPS_CONFIG_HEADER, so config.hpp selects
+# rtps/config_esp32.hpp via ESP_PLATFORM. The relaxed "host" / "host_large"
+# profiles override the profile header. These are capacity-only caps and do not
+# change any bytes on the wire.
+if(CONFIG_RTPS_LIMITS_PROFILE_HOST)
+  target_compile_definitions(${COMPONENT_LIB} PUBLIC RTPS_CONFIG_HEADER="rtps/config_desktop.hpp")
+elseif(CONFIG_RTPS_LIMITS_PROFILE_HOST_LARGE)
+  target_compile_definitions(${COMPONENT_LIB} PUBLIC RTPS_CONFIG_HEADER="rtps/config_host_large.hpp")
+endif()
+
+# Storage policy is orthogonal to the limits profile above: dynamic (heap,
+# grow-on-full) storage is an explicit ESP opt-in (default static) so a relaxed
+# limits profile never silently switches the MCU to heap-backed history. The
+# limits headers no longer define RTPS_STORAGE_DYNAMIC themselves; it is set here
+# on ESP and defaulted on in config.hpp for host/PC builds.
+if(CONFIG_RTPS_STORAGE_DYNAMIC)
+  target_compile_definitions(${COMPONENT_LIB} PUBLIC RTPS_STORAGE_DYNAMIC)
+endif()
+
+# Best-effort DATA_FRAG fragmentation is opt-in on ESP targets (Kconfig, default
+# off) so the MCU pays nothing for it by default. When enabled, define
+# RTPS_ENABLE_FRAGMENTATION (compiles the fragment send/reassembly paths) and the
+# reassembly cap RTPS_MAX_SAMPLE_SIZE (256 KB on the embedded profile). The
+# facade's max payload size rises to this when fragmentation is enabled.
+if(CONFIG_RTPS_ENABLE_FRAGMENTATION)
+  target_compile_definitions(${COMPONENT_LIB} PUBLIC
+    RTPS_ENABLE_FRAGMENTATION RTPS_MAX_SAMPLE_SIZE=262144)
+endif()
+
+# The RPC layer (services + actions) is compiled in by default; the facade
+# header defines RTPS_WITH_RPC unless RTPS_NO_RPC is set. When the Kconfig option
+# is turned OFF, define RTPS_NO_RPC so the whole services/actions surface (and its
+# std::thread/std::future use) is excluded, saving flash. ESP-only (this file is
+# the ESP-IDF component build); host builds always keep RPC on.
+if(NOT CONFIG_RTPS_ENABLE_RPC)
+  target_compile_definitions(${COMPONENT_LIB} PUBLIC RTPS_NO_RPC)
+endif()
+

components/rtps_embedded/Kconfig → components/rtps/Kconfig

@@ -1,11 +1,11 @@
-menu "RTPS (rtps_embedded)"
+menu "RTPS"
 
     choice RTPS_LIMITS_PROFILE
         prompt "RTPS static limits profile"
         default RTPS_LIMITS_PROFILE_EMBEDDED
         help
             Selects the compile-time capacity limits (profile header) used by the
-            rtps_embedded engine. Only the capacity caps differ between profiles.
+            rtps engine. Only the capacity caps differ between profiles.
             The storage MODEL (static vs dynamic) is a separate knob
             (RTPS_STORAGE_DYNAMIC below) and defaults to fully-static on ESP for
             every profile - selecting a relaxed profile does NOT enable heap