feat(rtps_embedded): ROS 2-interoperable services (RMI) + actions (AMI) + native protocol

.github/workflows/build.yml

@@ -213,6 +213,8 @@ 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

@@ -16,6 +16,16 @@ on:
       - ".github/workflows/rtps_interop.yml"
   workflow_dispatch:
 
+# Supersede in-progress runs: a new commit on the same PR (or the same branch for
+# a manual dispatch) cancels the earlier, now-stale interop run. Keyed by the
+# workflow (so it never cross-cancels other workflows) + the PR number, which is
+# globally unique - unlike the head branch name, two fork PRs can't collide on it.
+# Falls back to github.ref for workflow_dispatch (no PR number there). This
+# workflow never runs on push to main, so cancel-in-progress is always safe here.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   interop:
     runs-on: ubuntu-latest

components/rtps_embedded/CMakeLists.txt

@@ -54,3 +54,12 @@ if(CONFIG_RTPS_ENABLE_FRAGMENTATION)
     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

@@ -57,4 +57,15 @@ menu "RTPS (rtps_embedded)"
             bounded by RTPS_MAX_SAMPLE_SIZE (256 KB on the embedded profile) and
             the participant's max payload size rises accordingly.
 
+    config RTPS_ENABLE_RPC
+        bool "Enable RPC: services + actions (RMI/AMI)"
+        default y
+        help
+            Compiles in the request/reply (services) and goal (actions) layers of
+            the RtpsParticipant facade - both the ROS 2-interoperable
+            (add_service_*/add_action_*) and native (add_native_*) variants.
+            Disable to drop all of that code (and its std::thread/std::future use)
+            when the device only needs pub/sub, saving flash. Pure pub/sub is
+            unaffected either way. Default ON.
+
 endmenu

components/rtps_embedded/README.md

@@ -1,18 +1,77 @@
 # rtps_embedded
 
 ESPP component that integrates the [embeddedRTPS](https://github.com/embedded-software-laboratory/embeddedRTPS)
-RTPS/DDS stack into the ESPP ecosystem.  
-Any platform that can build ESPP — including ESP32, Linux, and desktop PCs —
-can use this component to discover and exchange typed messages with ROS 2 nodes
-or any other DDS participant on the same network using the standard RTPS wire
-protocol.
-
-The original embeddedRTPS library has hard dependencies on FreeRTOS and lwIP.
-`rtps_embedded` removes those dependencies by replacing all socket, task, and
-synchronisation calls with ESPP's platform-agnostic `UdpSocket`, `Task`, and
-`ThreadPool` primitives.  When built for ESP32, ESPP uses FreeRTOS and lwIP
-under the hood; on other platforms it uses the host OS equivalents — the RTPS
-code itself is unchanged in either case.
+RTPS/DDS stack into the ESPP ecosystem, behind an idiomatic `espp::RtpsParticipant`
+facade. Any platform that can build ESPP — ESP32, Linux, macOS, Windows — can use
+it to interoperate with **ROS 2** nodes (rmw_fastrtps) or any DDS participant on
+the network over the standard RTPS wire protocol.
+
+It provides three messaging patterns, all validated against live ROS 2:
+
+- **Pub/sub** — topic-based, best-effort or reliable (HEARTBEAT/ACKNACK).
+- **Services (RMI)** — request/reply with correlated responses.
+- **Actions (AMI)** — long-running goals with feedback, result, and cancellation.
+
+Each has a **typed** layer (reflectable structs, no manual bytes) and a
+**byte-level** layer. Services and actions come in a ROS 2-interoperable flavour
+and a lean **native** (espp ↔ espp) flavour.
+
+The upstream embeddedRTPS library hard-depends on FreeRTOS and lwIP;
+`rtps_embedded` removes those by routing all socket, task, and synchronisation
+through ESPP's platform-agnostic `UdpSocket`, `Task`, `ThreadPool`, and
+`SocketReactor`. On ESP32 those map to lwIP + FreeRTOS; elsewhere to the host OS.
+Micro-CDR is gone — (de)serialization uses ESPP's reflection-driven `cdr`.
+
+---
+
+## Quick-start (typed facade)
+
+```cpp
+#include "rtps_participant.hpp"
+#include "rtps_pubsub.hpp"   // typed Publisher<T> / Subscriber<T>
+#include "rtps_service.hpp"  // typed ServiceServer / ServiceClient
+#include "rtps_action.hpp"   // typed ActionServer  / ActionClient
+
+// Any reflectable struct is a message - fields map straight to CDR.
+struct StringMsg  { std::string data; };
+struct AddReq     { int64_t a, b; };
+struct AddResp    { int64_t sum; };
+
+espp::RtpsParticipant participant({.interface_address = "192.168.1.10"});
+participant.start();
+
+// Pub/sub
+espp::Publisher<StringMsg>  pub(participant, {.topic = "rt/chatter",
+                                              .type_name = "std_msgs::msg::dds_::String_",
+                                              .reliability = espp::RtpsParticipant::Reliability::RELIABLE});
+espp::Subscriber<StringMsg> sub(participant, {.topic = "rt/chatter",
+                                              .type_name = "std_msgs::msg::dds_::String_",
+                                              .on_message = [](const StringMsg &m) { /* use m.data */ }});
+pub.publish(StringMsg{"hello"});
+
+// Service (RMI) - ros2 service call /add_two_ints ... hits this server
+espp::ServiceServer<AddReq, AddResp> server(participant, {
+    .service = "/add_two_ints", .type_name = "example_interfaces::srv::dds_::AddTwoInts",
+    .handler = [](const AddReq &r) { return AddResp{r.a + r.b}; }});
+espp::ServiceClient<AddReq, AddResp> client(participant, {
+    .service = "/add_two_ints", .type_name = "example_interfaces::srv::dds_::AddTwoInts"});
+if (auto resp = client.call(AddReq{7, 35}, std::chrono::seconds(1))) { /* resp->sum == 42 */ }
+```
+
+For ROS 2 interop use ROS 2 naming: topic `rt/<name>`, type `<pkg>::msg::dds_::<Type>_`.
+The full request/reply + goal APIs (including the three client call styles and the
+native protocol) are documented in
+[`doc/en/protocols/rtps_rmi_ami.rst`](../../doc/en/protocols/rtps_rmi_ami.rst).
+
+### Byte-level API
+
+The typed wrappers are thin layers over `espp::RtpsParticipant`'s byte-level
+methods (`add_writer`/`add_reader`/`publish`, `add_service_server`/`_client`,
+`add_action_server`/`_client`, and the `add_native_*` variants), which take/return
+CDR-encapsulated `std::span<const uint8_t>`. Use those for dynamic types.
+
+Python bindings expose the same surface via the `espp` module (see
+[`python/rtps_rpc_demo.py`](../../python/rtps_rpc_demo.py)).
 
 ---
 
@@ -22,101 +81,47 @@ code itself is unchanged in either case.
 user code
     │
     ▼
-rtps::Domain          — routes packets to participants; owns discovery threads
-    │
-    ├── rtps::Participant   — groups writers and readers
-    │       ├── rtps::Writer  — publishes CacheChange samples
-    │       └── rtps::Reader  — delivers samples to a user callback
+espp::RtpsParticipant         — the facade: start()/stop(), add_writer/reader,
+    │                           publish, add_service_*/add_action_*
+    ▼
+rtps::Domain                  — routes packets to participants; owns discovery
     │
-    ├── rtps::ThreadPool    — espp::ThreadPool workers that drain the four
-    │                         incoming/outgoing meta/user traffic queues
+    ├── rtps::Participant      — groups writers and readers
+    │       ├── rtps::Writer   — publishes CacheChange samples
+    │       └── rtps::Reader   — delivers samples to a user callback
     │
-    └── rtps::EsppTransport — one espp::UdpSocket per open UDP port,
-                              each with its own receive task
+    └── rtps::EsppTransport    — the sole platform-specific adapter: one
+                                 espp::UdpSocket per UDP port, dispatched by an
+                                 espp::SocketReactor onto a shared espp::ThreadPool
+                                 (also used for async writer work)
 ```
 
-`EsppTransport` is the sole platform-specific adapter.  It wraps ESPP's
-`UdpSocket` and `Task`, which in turn map to:
-
 | Build target | Socket backend | Task backend |
 |---|---|---|
 | ESP32 | lwIP (via ESP-IDF) | FreeRTOS |
-| Linux / PC | POSIX sockets | `std::thread` |
-
----
-
-## Quick-start
-
-```cpp
-#include "rtps/entities/Domain.h"
-
-// 1. Construct the domain with the local interface IP.
-rtps::Domain domain(local_ip);
-
-// 2. Create a participant *before* completeInit().
-rtps::Participant *part = domain.createParticipant();
-
-// 3. Add user-defined writer and reader endpoints.
-rtps::Writer *writer = domain.createWriter(*part, "my/topic",
-                                           "std_msgs::msg::String", false);
-rtps::Reader *reader = domain.createReader(*part, "my/topic",
-                                           "std_msgs::msg::String", false);
-
-// 4. Register a receive callback on the reader.
-reader->registerCallback(
-    [](void *, const rtps::ReaderCacheChange &change) {
-        // process change.getData() / change.copyInto(...)
-    }, nullptr);
-
-// 5. Start discovery (SPDP/SEDP) and worker threads.
-domain.completeInit();
-
-// 6. Publish a sample.
-const char *payload = "hello";
-writer->newChange(rtps::ChangeKind_t::ALIVE,
-                  reinterpret_cast<const uint8_t *>(payload),
-                  static_cast<rtps::DataSize_t>(strlen(payload) + 1));
-```
+| Linux / macOS / PC | POSIX sockets | `std::thread` |
 
-> **Note**: `createParticipant()` **must** be called before `completeInit()`.
-> No new participants can be added after init is complete.
+Services/actions are pure library code over pub/sub — the only wire addition is
+a `related_sample_identity` inline QoS on service replies (for ROS 2 correlation).
 
 ---
 
 ## Configuration
 
-Two built-in config headers are provided. Select one by defining
-`RTPS_CONFIG_HEADER`, or let `include/rtps/config.h` pick automatically based
-on the build target.
+Capacity limits are chosen at build time by a **limits profile** header; storage
+policy, fragmentation, and the RPC layer are separate, independent knobs. On
+ESP32 these are ESP-IDF menuconfig options (`RTPS (rtps_embedded)`); on host they
+default via `include/rtps/config.hpp`.
 
-| Header | Target |
-|---|---|
-| [`include/rtps/config_esp32.h`](include/rtps/config_esp32.h) | ESP32 (ESP-IDF) |
-| [`include/rtps/config_desktop.h`](include/rtps/config_desktop.h) | Linux / PC |
-
-All tunable constants follow the same layout in both files:
-
-| Constant | Default | Description |
+| Knob | Options / default | Effect |
 |---|---|---|
-| `DOMAIN_ID` | 0 | RTPS domain number (0–230 with UDP) |
-| `MAX_NUM_PARTICIPANTS` | 1 | Participant pool size |
-| `NUM_STATEFUL_WRITERS` | 5 | User writer endpoint pool |
-| `NUM_STATEFUL_READERS` | 5 | User reader endpoint pool |
-| `NUM_STATELESS_WRITERS` | 5 | Discovery writer endpoint pool |
-| `NUM_STATELESS_READERS` | 5 | Discovery reader endpoint pool |
-| `NUM_WRITERS_PER_PARTICIPANT` | 10 | Max writers per participant |
-| `NUM_READERS_PER_PARTICIPANT` | 10 | Max readers per participant |
-| `HISTORY_SIZE_STATEFUL` | 10 | Per-endpoint history depth |
-| `THREAD_POOL_NUM_WRITERS` | 2 | Writer worker threads |
-| `THREAD_POOL_NUM_READERS` | 2 | Reader worker threads |
-| `THREAD_POOL_WRITER_STACKSIZE` | 4096 B | Writer task stack |
-| `THREAD_POOL_READER_STACKSIZE` | 6144 B | Reader / UDP-receive task stack |
-| `MAX_NUM_UDP_CONNECTIONS` | 10 | UDP socket pool size |
-| `SPDP_RESEND_PERIOD_MS` | 2000 | Discovery announce period |
-| `SF_WRITER_HB_PERIOD_MS` | 4000 | Reliable-writer heartbeat period |
-
-The `OVERALL_HEAP_SIZE` constant at the bottom of that file estimates the
-total stack RAM consumed by all internal tasks.
+| `RTPS_LIMITS_PROFILE` | `embedded` (default) / `host` / `host_large` | Compile-time endpoint/history capacity caps (`config_esp32.hpp` / `config_desktop.hpp` / `config_host_large.hpp`). Wire-neutral. |
+| `RTPS_STORAGE_DYNAMIC` | off on ESP32 / on host | Static `std::array` history (zero-heap, drop-oldest) vs heap-backed `std::deque` (grows). Orthogonal to the profile. |
+| `RTPS_ENABLE_FRAGMENTATION` | off on ESP32 / on host | DATA_FRAG for samples > ~64 KB (interoperates with FastDDS/ROS 2). |
+| `RTPS_ENABLE_RPC` | on (default) | Compile in services + actions (RMI/AMI). Disable to drop that code + its threads on a pure-pub/sub device. |
+
+The domain id, announcement/heartbeat periods, and pool sizes live in the profile
+headers.
 
 ---
 
@@ -125,35 +130,27 @@ total stack RAM consumed by all internal tasks.
 | Component | Purpose |
 |---|---|
 | `base_component` | ESPP base class with integrated `espp::Logger` |
-| `socket` | ESPP `UdpSocket` used by `EsppTransport` |
-| `task` | ESPP `Task` for per-port UDP receive loops |
-| `thread_pool` | ESPP `ThreadPool` for writer/reader workers |
-| `cdr` | CDR serialization helpers |
-
-These components abstract away all OS and network-stack details, so
-`rtps_embedded` itself has no direct dependency on FreeRTOS, lwIP, or any
-other platform library. Discovery (SPDP/SEDP) parameter-list serialization is
-built on the espp `cdr` component's stream primitives (see
-`include/rtps/utils/CdrBuffer.hpp`); the engine carries no vendored
-third-party code.
+| `socket` | `UdpSocket` + `SocketReactor` used by `EsppTransport` |
+| `task` | `espp::Task` / `espp::Timer` |
+| `thread_pool` | shared worker pool for receive dispatch + async writer work |
+| `cdr` | reflection-driven CDR/XCDR (de)serialization |
+
+The engine carries no vendored third-party code and has no direct dependency on
+FreeRTOS, lwIP, or any platform library.
 
 ---
 
 ## Example
 
-See [`example/`](example/) for a two-node **initiator / responder** demo.
-
-The same logic runs on any ESPP-supported platform. For ESP32, flash one board
-as *Initiator* and a second as *Responder* via menuconfig
-(`idf.py menuconfig → RTPS Example Configuration`). The initiator periodically
-publishes numbered request messages; the responder echoes each message back on
-the response topic.
+See [`example/`](example/) — an ESP32 (esp32-ethernet-kit) node that brings up a
+participant over Ethernet and exercises the typed APIs: a `Publisher`/`Subscriber`
+pair, a `ServiceServer` (`/add_two_ints`) + `ActionServer` (`/fibonacci`) a ROS 2
+client can drive, and a `ServiceClient` + `ActionClient`. A `menuconfig` option
+adds a second, self-testing participant. See [`example/README.md`](example/README.md).
 
-Key menuconfig options (ESP32 example):
+## Interop & tests
 
-| Option | Description |
-|---|---|
-| `RTPS_EXAMPLE_ROLE` | `Initiator` or `Responder` |
-| `RTPS_EXAMPLE_TOPIC_PREFIX` | Shared topic prefix (e.g. `espp/rtps_example`) |
-| `RTPS_EXAMPLE_PUBLISH_PERIOD_MS` | Initiator publish interval |
-| `ESP_WIFI_SSID` / `ESP_WIFI_PASSWORD` | Wi-Fi credentials |
+[`interop/`](interop/) runs a dockerised FastDDS / ROS 2 (Jazzy) matrix — golden
+byte-for-byte wire tests, in-process loopbacks (pub/sub, services, actions,
+native, typed), and live `ros2 service call` / `ros2 action send_goal` both
+directions. It is gated in CI (`.github/workflows/rtps_interop.yml`).