RTPS APIs

The rtps component is a cross-platform RTPS / DDS stack: it integrates the embeddedRTPS engine behind an idiomatic espp::RtpsParticipant facade so that any platform that can build ESPP — ESP32, Linux, macOS, Windows — can 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, each validated against live ROS 2 (Jazzy), and each with a typed layer (reflectable structs, no manual bytes) and a byte-level layer:

• 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.

Services and actions come in a ROS 2-interoperable flavour and a lean native (espp ↔ espp) flavour. The wire-format details of the RMI/AMI layers are in RTPS Services & Actions (RMI / AMI).

Note

The upstream embeddedRTPS library hard-depends on FreeRTOS and lwIP. This component removes those, routing all socket, task, and synchronisation through ESPP’s platform-agnostic UdpSocket, Task, ThreadPool, and SocketReactor (lwIP + FreeRTOS on ESP32, the host OS elsewhere). Micro-CDR is gone — (de)serialization uses ESPP’s reflection-driven cdr.

Architecture

The only platform-specific code is EsppTransport; everything above it is portable C++23. The espp:: facade is a thin, typed surface over the rtps:: engine.

        flowchart TD
  U["User code / ROS 2 peer"]

  subgraph facade["espp:: facade (typed + byte-level)"]
    PS["Publisher / Subscriber (typed)"]
    SVC["ServiceServer / ServiceClient"]
    ACT["ActionServer / ActionClient"]
    RP["espp::RtpsParticipant"]
    PS --> RP
    SVC --> RP
    ACT --> RP
  end

  subgraph engine["rtps:: engine (embeddedRTPS, de-vendored)"]
    DOM["rtps::Domain — packet routing + discovery"]
    PART["rtps::Participant"]
    WR["rtps::Writer (history + HEARTBEAT)"]
    RD["rtps::Reader (ACKNACK + delivery)"]
    DISC["SPDP + SEDP discovery agents"]
    DOM --> PART --> WR & RD
    DOM --> DISC
  end

  subgraph plat["platform adapter (the ONLY porting layer)"]
    TR["rtps::EsppTransport"]
    SOCK["espp::UdpSocket × N ports"]
    REACT["espp::SocketReactor → espp::ThreadPool (QosBand priority)"]
    CDR["espp::cdr (reflection CDR/XCDR)"]
    TR --> SOCK --> REACT
  end

  U --> facade --> engine --> plat
  WR -. serialize .-> CDR
  RD -. deserialize .-> CDR
    

Discovery Flow

RTPS separates metatraffic (discovery) from user traffic (samples). A participant announces itself with SPDP over multicast, then exchanges SEDP endpoint metadata (topic, type, reliability, locators) with each peer. Once a local writer and a remote reader (or vice-versa) match on topic + type, user data flows.

        sequenceDiagram
  participant A as espp participant
  participant MC as 239.255.0.1 (SPDP)
  participant B as ROS 2 / DDS peer
  A->>MC: SPDP DATA(GUID, locators, builtin endpoints, user_data)
  MC-->>B: multicast delivery
  B->>MC: SPDP DATA(its participant metadata)
  MC-->>A: multicast delivery
  A->>B: SEDP publication/subscription DATA(topic, type, QoS, unicast locator)
  B->>A: SEDP publication/subscription DATA
  Note over A,B: writer/reader match on topic + type → user data flows
    

Pub/Sub Reliability

A best-effort writer simply sends DATA submessages; lost samples are not recovered. A reliable writer keeps a history and periodically piggybacks a HEARTBEAT advertising its sequence-number range; the reader ACKNACKs what it is missing, and the writer retransmits. This is the state machine that makes the component interoperate with reliable ROS 2 QoS.

        sequenceDiagram
  participant W as Reliable Writer
  participant R as Reader
  W->>R: DATA (seq 1)
  W-xR: DATA (seq 2)
  W->>R: DATA (seq 3)
  Note over W,R: seq 2 was lost
  W->>R: HEARTBEAT (first=1, last=3)
  R->>W: ACKNACK (missing = 2)
  W->>R: DATA (seq 2) retransmit
  R->>W: ACKNACK (all received)
    

Large samples (> ~64 KB) are split into DATA_FRAG submessages when RTPS_ENABLE_FRAGMENTATION is on (default on host, Kconfig opt-in on ESP32); this interoperates with FastDDS both directions.

Services (RMI)

A service is two topics — a request and a reply — plus correlation. The client tags its request with a related_sample_identity inline QoS carrying its own reply-reader GUID; the server echoes {that GUID, the request's sequence number} on the reply, and the client matches the reply back to the pending call. This is exactly what rmw_fastrtps does, so an espp service appears in ros2 service list and answers ros2 service call.

        sequenceDiagram
  participant C as ServiceClient
  participant S as ServiceServer
  C->>S: request DATA (rq topic) + related_sample_identity{reply-reader GUID, seq=UNKNOWN}
  Note over S: handler(request) produces response
  S->>C: reply DATA (rr topic) + related_sample_identity{that GUID, request seq}
  Note over C: match on {own GUID, pending seq} → deliver reply
    

Clients offer three call styles: blocking call(), callback call_async(), and call_future() returning a std::future.

Actions (AMI)

An action adds no new wire primitive: it composes three services (send_goal / cancel_goal / get_result) and two topics (feedback / status) over the RMI + pub/sub layers. A goal moves through a small lifecycle the server drives and the client observes:

        stateDiagram-v2
  [*] --> ACCEPTED: send_goal (accepted)
  [*] --> REJECTED: send_goal (rejected)
  ACCEPTED --> EXECUTING: execute() starts
  EXECUTING --> EXECUTING: publish_feedback()
  EXECUTING --> SUCCEEDED: succeed(result)
  EXECUTING --> ABORTED: abort(result)
  EXECUTING --> CANCELED: cancel_goal + is_canceling() → canceled(result)
  SUCCEEDED --> [*]: get_result
  ABORTED --> [*]: get_result
  CANCELED --> [*]: get_result
  REJECTED --> [*]
    

Native protocol

For espp ↔ espp links that do not need ROS 2 interop, a lean native protocol trades interop for simplicity: services correlate with a 20-byte in-band header (no inline QoS, no rq/rr mangling) on es_rq / es_rr topics, and a native action is just two services + one topic (a goal service, a cancel service, and one feedback topic that also carries the terminal result: 5 topics / 5 RTPS endpoints per side, vs the ROS 2 action’s three services + two topics = 8). Same client ergonomics. See RTPS Services & Actions (RMI / AMI) for the byte layout.

Ports and Channels

The component follows the standard UDPv4 RTPS port mapping formula:

Channel	Formula	Domain 0, participant 0
Metatraffic multicast	7400 + 250 * domain + 0	7400
Metatraffic unicast	7400 + 250 * domain + 10 + 2 * participant	7410
User multicast	7400 + 250 * domain + 1	7401
User unicast	7400 + 250 * domain + 11 + 2 * participant	7411
Dedicated endpoint (prioritized)	7400 + 250 * domain + 100 + n	7500, 7501, …

Every channel is one espp::UdpSocket registered on the transport’s espp::SocketReactor at a priority band (espp::QosBand). By default the metatraffic channels (SPDP multicast + SEDP unicast) run at QosBand::High so discovery dispatch overtakes queued user-traffic handling under load, and the shared user channels run at Normal; both are configurable (Config::metatraffic_band / Config::user_traffic_band).

Per-endpoint priority (dedicated ports)

All user traffic for a participant normally shares ONE user-unicast port, so per-socket priority alone cannot distinguish endpoints. An endpoint (writer or reader) configured with a non-default band — or a dscp marking, which is per-socket — is therefore granted its own dedicated unicast port:

• the port is allocated deterministically from the domain’s RTPS port block at offset 100 (7400 + 250*domain + 100 + n). Each allocation probes at most 16 consecutive candidates (a reuse-disabled bind, so ports taken by other processes fail loudly) starting at an advancing cursor — if the whole window is occupied, that endpoint falls back to the shared user port (with a warning) and the cursor advances past the window, so the next allocation probes fresh ports rather than one request scanning the entire 100..249 range. The standard RTPS offsets stay below 100 only for participant ids 0–44, so participant creation enforces that cap while dedicated ports are enabled;

• its socket is registered on the reactor at the endpoint’s band and optionally DSCP-marked (espp::Dscp, e.g. Dscp::Ef) — the endpoint also sends from this socket, so the marking applies to its outgoing traffic;

• the endpoint’s SEDP announcement carries the dedicated port as its standard per-endpoint unicast locator (PID_UNICAST_LOCATOR), so FastDDS / ROS 2 peers send that endpoint’s traffic straight to the prioritized socket. The wire format is unchanged — only the announced port value differs.

Dedicated ports are rationed (Config::max_prioritized_endpoint_ports, default 4): each one consumes a UDP socket/fd, and lwIP on ESP32 defaults to ~10 sockets total of which the participant already uses 4. When the ration is exhausted (or Config::enable_dedicated_endpoint_ports is false), a banded endpoint logs a warning and falls back to the shared port; banded readers then use deferred banded dispatch — samples are queued (bounded) and the on_sample callback is re-submitted to the transport’s worker pool at the reader’s band, one in-flight delivery per reader, preserving order. Endpoints left at QosBand::Normal keep the exact pre-band inline delivery path.

Configuration

Capacity limits are chosen at build time by a limits profile header; storage policy, fragmentation, and the RPC layer are separate, independent knobs (ESP-IDF menuconfig under RTPS on ESP32; include/rtps/config.hpp defaults on host).

Knob	Options / default	Effect
RTPS_LIMITS_PROFILE	embedded (ESP32 default) / host (host default) / host_large	Compile-time endpoint/history capacity caps. 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 and its threads on a pure-pub/sub device.

Relevant Specifications

Specification	Why it matters here
OMG DDSI-RTPS 2.3	Primary wire-level reference for RTPS headers, submessages (DATA, HEARTBEAT, ACKNACK, DATA_FRAG), SPDP, SEDP, locator encoding, GUIDs, and the UDP port mapping used by this component.
OMG DDS 1.4	The participant / reader / writer / topic / QoS model that RTPS carries.

Example

The RTPS Example page shows 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) and ActionServer (/fibonacci) a ROS 2 client can drive, and a ServiceClient / ActionClient.

• RTPS Example
  • How to use example
    • Configure
    • Build and Flash
    • Talk to it
  • Expected Output

API Reference

Header File

• components/rtps/include/rtps_participant.hpp

Macros

RTPS_WITH_RPC

Classes

class RtpsParticipant : public espp::BaseComponent

RTPS/DDS participant for pub/sub interop with FastDDS and ROS 2.

An espp-idiomatic facade over the embeddedRTPS engine (the FastDDS/ROS 2 interop-proven RTPS implementation vendored in components/rtps). One RtpsParticipant owns one RTPS domain participant: create it with a Config, start() it, then add writers/readers and publish CDR-encapsulated samples. Samples arriving on readers are delivered via the on_sample callback as CDR-encapsulated payload bytes (use the reflection-driven `cdr` component - cdr::serialize / cdr::deserialize - to (de)serialize them).

For ROS 2 interop, use ROS 2 naming conventions: topic “rt/<name>” and type “<pkg>::msg::dds_::<Type>_” (e.g. topic “rt/chatter” with type “std_msgs::msg::dds_::String_” matches a ROS 2 std_msgs/String subscriber on /chatter).

Engine limitations that still apply (see components/rtps/REFACTOR_PLAN.md): domain id is fixed at compile time (Config::DOMAIN_ID, default 0), announcement/heartbeat periods are compile-time constants, and endpoint counts are bounded by the engine’s pools. Multiple RtpsParticipants per process/host work - each probes forward to free unicast ports.

Priority scheduling: transport channels dispatch at espp::QosBand bands (metatraffic High by default - Config::metatraffic_band; user traffic Normal). Endpoints get per-endpoint priority via WriterConfig::band / ReaderConfig::band (and ServiceConfig / ActionConfig): a banded endpoint is granted a dedicated, band-scheduled (optionally DSCP-marked) unicast port, announced to peers via its SEDP unicast locator; when the dedicated-port ration (Config::max_prioritized_endpoint_ports) is exhausted, banded readers fall back to deferred banded dispatch. See the component README’s “Priority scheduling” section.

RtpsParticipant Example

  espp::RtpsParticipant participant({
      .interface_address = interface_address,
      .on_publisher_matched = [&]() { logger.info("publisher matched a remote reader"); },
      .on_subscriber_matched = [&]() { logger.info("subscriber matched a remote writer"); },
      .log_level = espp::Logger::Verbosity::INFO,
  });
  if (!participant.start()) {
    logger.error("Failed to start the RTPS participant");
    return;
  }

Full Example (network bring-up + typed pub/sub + RPC)

  // Bring up Ethernet (DHCP server on 192.168.4.1/24 so a directly-attached PC
  // gets an address); any espp network interface works - the RTPS participant
  // only needs the interface's IPv4 address.
  auto &board = espp::Esp32EthernetKit::get();
  bool eth_ok = board.initialize_ethernet({
      .mode = espp::Esp32EthernetKit::DhcpMode::SERVER,
      .on_link_up = [&]() { logger.info("Ethernet link up"); },
      .on_link_down = [&]() { logger.warn("Ethernet link down"); },
  });
  if (!eth_ok) {
    logger.error("Ethernet initialization failed");
    return;
  }
  logger.info("Waiting for Ethernet link...");
  while (!board.is_ethernet_connected()) {
    std::this_thread::sleep_for(100ms);
  }
  auto eth_ip = board.ethernet_ip();
  const std::string interface_address =
      fmt::format("{}.{}.{}.{}", esp_ip4_addr1_16(&eth_ip), esp_ip4_addr2_16(&eth_ip),
                  esp_ip4_addr3_16(&eth_ip), esp_ip4_addr4_16(&eth_ip));
  logger.info("Ethernet up, IP {}", interface_address);

  // RTPS/DDS participant (embeddedRTPS engine behind the espp facade). The
  // topics pair with the FastDDS host peer in example/pc/host_pubsub.cpp; for
  // ROS 2 instead, use topic "rt/<name>" with type "<pkg>::msg::dds_::<Type>_"
  // (e.g. "rt/chatter" + "std_msgs::msg::dds_::String_").
  constexpr const char *pub_topic = "mcu_to_pc";
  constexpr const char *sub_topic = "pc_to_mcu";
  constexpr const char *type_name = "std_msgs::msg::String";

  // Automatic locals: they RAII-clean up in reverse order on any early return
  // (subscriber/publisher stop referencing the participant before it is
  // destroyed), and the trailing while(true) keeps them alive in normal use.
  espp::RtpsParticipant participant({
      .interface_address = interface_address,
      .on_publisher_matched = [&]() { logger.info("publisher matched a remote reader"); },
      .on_subscriber_matched = [&]() { logger.info("subscriber matched a remote writer"); },
      .log_level = espp::Logger::Verbosity::INFO,
  });
  if (!participant.start()) {
    logger.error("Failed to start the RTPS participant");
    return;
  }

  // Typed reliable publisher: publish StringMsg structs directly (HEARTBEAT/
  // ACKNACK-acknowledged, retransmitted to matched readers). No manual CDR.
  using Reliability = espp::RtpsParticipant::Reliability;
  espp::Publisher<StringMsg> publisher(participant, {
                                                        .topic = pub_topic,
                                                        .type_name = type_name,
                                                        .reliability = Reliability::RELIABLE,
                                                    });
  // Typed subscriber: receive StringMsg structs directly.
  espp::Subscriber<StringMsg> subscriber(
      participant, {
                       .topic = sub_topic,
                       .type_name = type_name,
                       .on_message = [&](const StringMsg &msg) { logger.info("rx: {}", msg.data); },
                   });
  if (!publisher.is_valid() || !subscriber.is_valid()) {
    logger.error("Failed to create the typed publisher/subscriber");
    return;
  }

  // Publish a counter periodically via the typed publisher.
  uint32_t counter = 0;
  espp::Timer publish_timer({
      .name = "rtps_pub",
      .period = std::chrono::milliseconds(CONFIG_RTPS_EXAMPLE_ANNOUNCE_PERIOD_MS),
      .callback =
          [&]() {
            if (publisher.publish(StringMsg{fmt::format("msg {}", counter++)})) {
              logger.info("tx: msg {}", counter - 1);
            } else {
              logger.warn("tx dropped (history full)");
            }
            return false; // keep the timer running
          },
      .log_level = espp::Logger::Verbosity::WARN,
  });
  logger.info("started: pub='{}' sub='{}' type='{}'", pub_topic, sub_topic, type_name);

#ifdef RTPS_WITH_RPC
  // Typed service (RMI) server: a ROS 2 client can `ros2 service call
  // /add_two_ints example_interfaces/srv/AddTwoInts "{a: 7, b: 35}"` and get 42.
  // No manual CDR - the reflectable AddReq/AddResp structs are (de)serialized for
  // us. (Compiled out when CONFIG_RTPS_ENABLE_RPC is disabled.)
  espp::ServiceServer<AddReq, AddResp> add_service(
      participant, {
                       .service = "/add_two_ints",
                       .type_name = "example_interfaces::srv::dds_::AddTwoInts",
                       .handler =
                           [&](const AddReq &r) {
                             logger.info("service add_two_ints: {} + {} = {}", r.a, r.b, r.a + r.b);
                             return AddResp{r.a + r.b};
                           },
                   });

  // Typed action (AMI) server: a ROS 2 client can `ros2 action send_goal
  // /fibonacci example_interfaces/action/Fibonacci "{order: 5}"` and receive
  // feedback + the [0,1,1,2,3,5] result. execute() runs on its own thread.
  espp::ActionServer<FibGoal, FibSeq, FibSeq> fib_action(
      participant, {
                       .action = "/fibonacci",
                       .type_name = "example_interfaces::action::dds_::Fibonacci",
                       .on_goal = [&](const FibGoal &g) { return g.order > 0; },
                       .execute =
                           [&](auto &h) {
                             const int32_t order = h.goal().order;
                             std::vector<int32_t> seq{0, 1};
                             for (int32_t i = 1; i < order; ++i) {
                               seq.push_back(seq[i] + seq[i - 1]);
                               h.publish_feedback(FibSeq{seq});
                               std::this_thread::sleep_for(200ms);
                             }
                             h.succeed(FibSeq{seq});
                             logger.info("action fibonacci({}) done", order);
                           },
                   });
  if (!add_service.is_valid() || !fib_action.is_valid()) {
    logger.error("Failed to create the typed service/action servers");
    return;
  }
  logger.info("service '/add_two_ints' + action '/fibonacci' ready");

  // Also demonstrate the CLIENT side on-device: a typed service client + action
  // client that call services a peer hosts ("/peer_add_two_ints", "/peer_fib").
  // Run a ROS 2 / rclpy server (or another espp device) for those names to see a
  // full round-trip; until then the calls simply time out (logged), which still
  // exercises the client API on-target. (Calling this device's OWN services is
  // not possible - a participant filters out its own messages.)
  espp::ServiceClient<AddReq, AddResp> add_client(
      participant,
      {.service = "/peer_add_two_ints", .type_name = "example_interfaces::srv::dds_::AddTwoInts"});
  espp::ActionClient<FibGoal, FibSeq, FibSeq> fib_client(
      participant,
      {.action = "/peer_fib", .type_name = "example_interfaces::action::dds_::Fibonacci"});

  // Only one action goal in flight at a time: without a peer the goal never
  // completes, so re-sending on every tick would leak a pending goal each time.
  // The service call() below self-cleans on its 1s timeout, so it can run freely.
  std::atomic<bool> fib_in_flight{false};
  espp::Timer rpc_client_timer({
      .name = "rtps_rpc_client",
      .period = 5s,
      .callback =
          [&]() {
            // Typed blocking service call (RMI).
            if (auto resp = add_client.call(AddReq{20, 22}, 1s)) {
              logger.info("[client] /peer_add_two_ints(20,22) = {}", resp->sum);
            } else {
              logger.info("[client] /peer_add_two_ints: no reply (peer serving it?)");
            }
            // Typed action goal (AMI) with typed feedback + result. Skip if the
            // previous goal has not finished (e.g. no peer is serving it).
            if (!fib_in_flight.exchange(true)) {
              fib_client.send_goal(
                  FibGoal{5}, [&](const FibSeq &) { /* per-feedback */ },
                  [&](espp::GoalStatus status, const FibSeq &res) {
                    logger.info("[client] /peer_fib result: status={} len={}",
                                static_cast<int>(status), res.sequence.size());
                    fib_in_flight.store(false);
                  });
            }
            return false; // keep the timer running
          },
      .log_level = espp::Logger::Verbosity::WARN,
  });
  if (!add_client.is_valid() || !fib_client.is_valid()) {
    logger.error("Failed to create the typed service/action clients");
    return;
  }
  logger.info("client for '/peer_add_two_ints' + '/peer_fib' running");

#if CONFIG_RTPS_EXAMPLE_SECOND_PARTICIPANT
  // Purely additive on-device SELF-TEST (Kconfig, default off): a SECOND
  // participant with its own service + action clients that call THIS device's own
  // /add_two_ints and /fibonacci servers, for a full local round-trip (a
  // participant filters out its own messages, so the loopback needs a distinct
  // participant). This roughly doubles the RTPS engine RAM - only enable on a
  // target with headroom (e.g. PSRAM).
  espp::RtpsParticipant selftest_participant({
      .interface_address = interface_address,
      .log_level = espp::Logger::Verbosity::WARN,
  });
  if (!selftest_participant.start()) {
    logger.error("Failed to start the self-test participant");
    return;
  }
  espp::ServiceClient<AddReq, AddResp> selftest_add_client(
      selftest_participant,
      {.service = "/add_two_ints", .type_name = "example_interfaces::srv::dds_::AddTwoInts"});
  espp::ActionClient<FibGoal, FibSeq, FibSeq> selftest_fib_client(
      selftest_participant,
      {.action = "/fibonacci", .type_name = "example_interfaces::action::dds_::Fibonacci"});
  espp::Timer selftest_timer({
      .name = "rtps_selftest",
      .period = 5s,
      .callback =
          [&]() {
            if (auto resp = selftest_add_client.call(AddReq{20, 22}, 2s)) {
              logger.info("[self-test] /add_two_ints(20,22) = {} ({})", resp->sum,
                          resp->sum == 42 ? "PASS" : "FAIL");
            } else {
              logger.warn("[self-test] /add_two_ints: no reply");
            }
            selftest_fib_client.send_goal(
                FibGoal{5}, [&](const FibSeq &) {},
                [&](espp::GoalStatus status, const FibSeq &res) {
                  const std::vector<int32_t> expected{0, 1, 1, 2, 3, 5};
                  const bool ok = status == espp::GoalStatus::SUCCEEDED && res.sequence == expected;
                  logger.info("[self-test] /fibonacci(5) len={} ({})", res.sequence.size(),
                              ok ? "PASS" : "FAIL");
                });
            return false; // keep the timer running
          },
      .log_level = espp::Logger::Verbosity::WARN,
  });
  if (!selftest_add_client.is_valid() || !selftest_fib_client.is_valid()) {
    logger.error("Failed to create the self-test clients");
    return;
  }
  logger.info("self-test participant round-tripping the local service + action");
#endif // CONFIG_RTPS_EXAMPLE_SECOND_PARTICIPANT
#endif // RTPS_WITH_RPC

Public Types

enum class Reliability

Reliability QoS for a writer or reader.

Values:

enumerator BEST_EFFORT

Fire-and-forget delivery (stateless endpoint).

enumerator RELIABLE

HEARTBEAT/ACKNACK acknowledged delivery (stateful endpoint).

using sample_callback_t = std::function<void(std::span<const uint8_t> cdr_payload)>

Callback for samples received on a reader. The span holds the CDR-encapsulated payload (4-byte encapsulation header + CDR body) and is only valid for the duration of the callback; copy it if you keep it.

Note

Runs on an engine worker thread - return quickly, do not block.

using matched_callback_t = std::function<void()>

Callback invoked when a remote endpoint matches one of this participant’s writers (publisher matched) or readers (subscriber matched).

Note

Runs on an engine worker thread - return quickly, do not block.

using service_handler_t = std::function<std::vector<uint8_t>(std::span<const uint8_t> request)>

Handler for a service server: given a CDR-encapsulated request, return the CDR-encapsulated reply. Runs on an engine worker thread - return promptly.

using service_deferred_handler_t = std::function<void(std::span<const uint8_t> request, ServiceResponder responder)>

Deferred service handler: invoked with the request and a responder. The handler may call responder.reply() immediately or store the responder and reply later (from any thread). Unlike service_handler_t this never blocks a worker waiting for a slow response.

using GoalId = std::array<uint8_t, 16>

16-byte action goal id (unique_identifier_msgs/UUID).

using action_goal_callback_t = std::function<bool(const GoalId &goal_id, std::span<const uint8_t> goal)>

Called when a goal arrives; return true to accept, false to reject.

using action_execute_callback_t = std::function<void(ActionGoalHandle handle)>

Called (on its own thread) to run an accepted goal to completion.

using action_cancel_callback_t = std::function<bool(const GoalId &goal_id)>

Called when a cancel is requested for a goal; return true to accept.

using native_cancel_callback_t = std::function<bool(uint32_t goal_handle)>

Cancel policy: return true to accept a cancel request for the goal_handle (the execute callback then observes is_canceling()). Default (nullptr) accepts every cancel.

Public Functions

explicit RtpsParticipant(const Config &config)

Construct the participant (does not open sockets; see start()).

Parameters:

config – The participant configuration.

~RtpsParticipant()

Stops the participant (see stop()).

bool start()

Start the participant: bring up the RTPS transport and begin SPDP/SEDP discovery. Writers and readers can only be added after a successful start().

Returns:

True on success (false if already started or bring-up failed).

void stop()

Stop the participant and its discovery/transport threads. Registered callbacks will not be invoked after stop() returns.

inline bool is_started() const

Returns:

True if the participant has been started and not stopped.

bool add_writer(const WriterConfig &config)

Add a publishing endpoint.

Parameters:

config – The writer configuration.

Returns:

True on success (false when not started, on duplicate topic, or when the engine’s writer pool is exhausted).

bool add_reader(const ReaderConfig &config)

Add a subscribing endpoint.

Parameters:

config – The reader configuration.

Returns:

True on success (false when not started or when the engine’s reader pool is exhausted).

bool publish(std::string_view topic, std::span<const uint8_t> cdr_payload)

Publish a CDR-encapsulated sample on a topic previously registered with add_writer().

Parameters:

• topic – The topic name used in add_writer().

• cdr_payload – The CDR-encapsulated sample (4-byte encapsulation header + CDR body); copied into the writer’s history. Must not exceed max_payload_size bytes.

Returns:

True if the sample was accepted into the writer history; false if the payload exceeds max_payload_size (see that constant).

bool add_service_server(const ServiceConfig &config, service_handler_t handler)

Add a service server. The handler is invoked for each request; its return value is sent back as the reply, correlated to the requesting client.

Returns:

True on success (false when not started or endpoint creation fails).

bool add_service_server_deferred(const ServiceConfig &config, service_deferred_handler_t handler)

Add a service server that replies asynchronously via a ServiceResponder. Use this when the response may not be ready when the request arrives (e.g. an action’s get_result).

Returns:

True on success.

std::shared_ptr<ServiceClient> add_service_client(const ServiceConfig &config)

Add a service client for calling a service.

Returns:

A client handle, or nullptr on failure (not started / endpoint creation failed). Owned by the participant; valid until stop().

bool add_action_server(const ActionConfig &config, action_goal_callback_t on_goal, action_execute_callback_t execute, action_cancel_callback_t on_cancel = nullptr)

Add an action server. on_goal decides acceptance; execute runs each accepted goal on its own thread; on_cancel (optional) accepts/rejects cancellations.

Returns:

True on success (false when not started or endpoint creation fails).

std::shared_ptr<ActionClient> add_action_client(const ActionConfig &config)

Add an action client.

Returns:

A handle, or nullptr on failure.

bool add_native_service_server(const ServiceConfig &config, service_handler_t handler)

Add a native (espp<->espp) service server.

Returns:

True on success.

std::shared_ptr<NativeServiceClient> add_native_service_client(const ServiceConfig &config)

Add a native (espp<->espp) service client.

Returns:

A handle, or nullptr.

bool add_native_action_server(const ActionConfig &config, native_goal_callback_t on_goal, native_execute_callback_t execute, native_cancel_callback_t on_cancel = nullptr)

Add a native action server. on_goal accepts/rejects; execute runs each accepted goal on its own thread; on_cancel (optional) gates cancel requests.

Returns:

True on success.

std::shared_ptr<NativeActionClient> add_native_action_client(const ActionConfig &config)

Add a native action client.

Returns:

A handle, or nullptr.

inline const std::string &get_name() const

Get the name of the component

Note

This is the tag of the logger

Returns:

A const reference to the name of the component

inline void set_log_tag(const std::string_view &tag)

Set the tag for the logger

Parameters:

tag – The tag to use for the logger

inline espp::Logger::Verbosity get_log_level() const

Get the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Returns:

The verbosity level of the logger

inline void set_log_level(espp::Logger::Verbosity level)

Set the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Parameters:

level – The verbosity level to use for the logger

inline void set_log_verbosity(espp::Logger::Verbosity level)

Set the log verbosity for the logger

See also

set_log_level

See also

Logger::Verbosity

See also

Logger::set_verbosity

Note

This is a convenience method that calls set_log_level

Parameters:

level – The verbosity level to use for the logger

inline espp::Logger::Verbosity get_log_verbosity() const

Get the log verbosity for the logger

See also

get_log_level

See also

Logger::Verbosity

See also

Logger::get_verbosity

Note

This is a convenience method that calls get_log_level

Returns:

The verbosity level of the logger

inline void set_log_rate_limit(std::chrono::duration<float> rate_limit)

Set the rate limit for the logger

See also

Logger::set_rate_limit

Note

Only calls to the logger that have _rate_limit suffix will be rate limited

Parameters:

rate_limit – The rate limit to use for the logger

Public Static Attributes

static constexpr std::size_t max_payload_size = 65451

Maximum size of a single published CDR payload, in bytes.

When fragmentation is compiled in (RTPS_ENABLE_FRAGMENTATION - always on host, opt-in on ESP32) this is the large-sample reassembly cap (RTPS_MAX_SAMPLE_SIZE: 8 MB host, 256 KB ESP32): samples above one DATA submessage are split into DATA_FRAG submessages and reassembled by the peer. When fragmentation is compiled out (ESP32 default) it is bounded by the RTPS wire format instead: a DATA submessage’s length field (octetsToNextHeader) is 16-bit, so one unfragmented sample cannot exceed 65535 bytes, and publish() rejects larger samples rather than truncating them.

class ActionClient

Client handle for calling an action. Obtain from add_action_client().

Public Types

using feedback_callback_t = std::function<void(std::span<const uint8_t> feedback)>

CDR-encapsulated feedback for an in-progress goal.

using result_callback_t = std::function<void(int8_t status, std::span<const uint8_t> result)>

Terminal result: the GoalStatus value + the CDR-encapsulated result.

Public Functions

std::optional<GoalId> send_goal(std::span<const uint8_t> goal, feedback_callback_t on_feedback, result_callback_t on_result)

Send a goal. on_feedback is invoked for each feedback message; on_result once when the goal terminates (or is rejected, with an empty result).

Returns:

The generated goal id, or std::nullopt on failure.

bool cancel_goal(const GoalId &goal_id)

Request cancellation of a previously sent goal.

struct ActionConfig

Configuration for an action server or client.

Public Members

std::string action

ROS 2 action name, e.g. “/fibonacci”.

std::string type_name

Base DDS action type, e.g. “example_interfaces::action::dds_::Fibonacci”.

espp::QosBand band = {espp::QosBand::Normal}

Priority band inherited by ALL of the action’s underlying endpoints: the send_goal/cancel_goal/get_result service endpoints and the feedback/status topic endpoints (a ROS action server is ~8 endpoints, a client ~7 - far more than the default dedicated-port ration of Config::max_prioritized_endpoint_ports, so most of a banded action’s endpoints will use the shared port; readers there get deferred banded dispatch. Raise the cap if you want dedicated ports for a whole action.) Native actions inherit it on their ~3 endpoints likewise.

std::optional<espp::Dscp> dscp = {}

Optional DSCP code point for the endpoints’ dedicated sockets.

class ActionGoalHandle

Server-side handle to a running goal, passed to the execute callback (which runs on its own thread). Publish feedback and terminate the goal through it.

Public Functions

std::span<const uint8_t> goal() const

The CDR-encapsulated goal payload.

void publish_feedback(std::span<const uint8_t> feedback) const

Publish a CDR-encapsulated feedback message for this goal.

void succeed(std::span<const uint8_t> result) const

Terminate the goal SUCCEEDED/ABORTED/CANCELED with a CDR result payload.

bool is_canceling() const

True if a cancel has been requested for this goal.

struct Config

Configuration for the participant.

Public Members

std::string interface_address = {}

IPv4 address of the network interface to use. On the host, leave empty to auto-detect the first non-loopback IPv4 interface. On ESP targets it must be set explicitly (e.g. from the WiFi/Ethernet netif IP).

matched_callback_t on_publisher_matched = {nullptr}

A writer gained a remote reader.

matched_callback_t on_subscriber_matched = {nullptr}

A reader gained a remote writer.

Logger::Verbosity log_level = {Logger::Verbosity::WARN}

Facade log verbosity.

espp::QosBand metatraffic_band = {espp::QosBand::High}

Priority band for the metatraffic (SPDP/SEDP discovery) channels. High by default so discovery dispatch stays responsive when user traffic backs the worker pool up; set to QosBand::Normal for the exact pre-band behavior.

espp::QosBand user_traffic_band = {espp::QosBand::Normal}

Priority band for the shared user-traffic channels (user unicast + user multicast). Normal by default (pre-band behavior).

bool enable_dedicated_endpoint_ports = {true}

Allow endpoints with a non-Normal band (or a dscp) to get a dedicated unicast port (see WriterConfig::band / ReaderConfig::band). Disable to force every banded endpoint onto the shared user port (readers then use deferred banded dispatch).

uint8_t max_prioritized_endpoint_ports = {4}

Cap on dedicated endpoint ports. Each consumes one UDP socket/fd - on ESP32, lwIP’s CONFIG_LWIP_MAX_SOCKETS defaults to ~10 total and the participant already uses 4 - so dedicated ports are deliberately rationed. When exhausted, further banded endpoints log a warning and fall back to the shared port (readers: deferred banded dispatch). The cap is a TRUE fd bound: a released endpoint’s socket counts against it until its fd actually closes (normally immediate).

class NativeActionClient

Client handle for a native action.

Public Types

using result_callback_t = std::function<void(uint8_t status, std::span<const uint8_t> result)>

Terminal result: the NativeGoalStatus value + the CDR result payload.

using accepted_callback_t = std::function<void(uint32_t goal_handle)>

Invoked once when the server accepts the goal, with the server-assigned goal_handle - keep it to cancel_goal() the goal later.

Public Functions

bool send_goal(std::span<const uint8_t> goal, feedback_callback_t on_feedback, result_callback_t on_result, accepted_callback_t on_accepted = nullptr)

Send a goal; on_feedback per feedback message, on_result once at the end, on_accepted (optional) with the assigned goal_handle when accepted.

Returns:

True if the goal was queued.

bool cancel_goal(uint32_t goal_handle)

Request cancellation of a previously accepted goal by its goal_handle (from on_accepted).

Returns:

True if the cancel request was queued.

class NativeGoalHandle

Server-side handle to a running native goal (passed to the execute callback, which runs on its own thread).

Public Functions

void canceled(std::span<const uint8_t> result) const

Terminate the goal CANCELED (in response to a cancel request).

bool is_canceling() const

True if the client has requested cancellation of this goal (and the server’s on_cancel, if any, accepted it). A long-running execute callback should poll this and wind down - calling canceled()/abort() - when set.

class NativeServiceClient

Client handle for a native service (see add_native_service_client()).

Public Functions

bool call_async(std::span<const uint8_t> request, reply_callback_t on_reply)

Send a request; invoke on_reply when the correlated reply arrives.

std::optional<std::vector<uint8_t>> call(std::span<const uint8_t> request, std::chrono::milliseconds timeout)

Send a request and block for the reply (std::nullopt on timeout/failure).

std::future<std::optional<std::vector<uint8_t>>> call_future(std::span<const uint8_t> request)

Send a request and return a future for the reply.

struct ReaderConfig

Configuration for a reader (subscribing endpoint).

Public Members

std::string topic

DDS topic name (e.g. “rt/chatter” for ROS 2).

std::string type_name

DDS type name (e.g. “std_msgs::msg::dds_::String_”).

Reliability reliability = {Reliability::BEST_EFFORT}

Reliability QoS.

sample_callback_t on_sample = {nullptr}

Called for each received sample.

espp::QosBand band = {espp::QosBand::Normal}

Priority band for this reader’s endpoint (see espp::QosBand). A non-Normal band (or a set dscp) requests a DEDICATED unicast port, announced to peers via the endpoint’s SEDP unicast locator (standard DDS, honored by FastDDS/ROS 2), so this reader’s samples arrive on their own socket and are dispatched at this band ahead of Normal traffic. Rationed - see Config::max_prioritized_endpoint_ports; when no dedicated port is available (or dedicated ports are disabled) the reader falls back to DEFERRED banded dispatch: its on_sample runs from a bounded per-reader queue re-submitted to the transport pool at this band (ordering preserved, one in-flight callback per reader) instead of inline on the shared-port receive worker.

std::optional<espp::Dscp> dscp = {}

Optional DSCP code point marking the traffic this reader SENDS (its ACKNACKs, when reliable). Requires a dedicated port; ignored when none could be allocated.

class ServiceClient

Client handle for calling a service. Obtain one from add_service_client(); it stays valid until the participant is stopped/destroyed.

Public Types

using reply_callback_t = std::function<void(std::span<const uint8_t> reply)>

Callback delivering a CDR-encapsulated reply for a call_async() request.

Public Functions

bool call_async(std::span<const uint8_t> request, reply_callback_t on_reply)

Send a request and invoke on_reply when the correlated reply arrives.

Returns:

False if the participant is not started or the request could not be queued. The callback runs on an engine worker thread.

std::optional<std::vector<uint8_t>> call(std::span<const uint8_t> request, std::chrono::milliseconds timeout)

Send a request and block until the correlated reply arrives or timeout.

Returns:

The CDR-encapsulated reply, or std::nullopt on timeout/failure. Do not call from within an engine callback (it would deadlock).

std::future<std::optional<std::vector<uint8_t>>> call_future(std::span<const uint8_t> request)

Send a request and return a future that becomes ready with the correlated reply (or std::nullopt if the request could not be queued). The future never blocks a worker thread; wait on it (or wait_for a timeout) from the caller. A pending request without a reply leaves the future unfulfilled until the participant stops.

struct ServiceConfig

Configuration for a service server or client.

Public Members

std::string service

ROS 2 service name, e.g. “/add_two_ints”.

std::string type_name

Base DDS service type, e.g. “example_interfaces::srv::dds_::AddTwoInts”. The _Request_/_Response_ suffixes are derived internally.

espp::QosBand band = {espp::QosBand::Normal}

Priority band applied to BOTH of the service’s endpoints: for a server its request reader + reply writer, for a client its request writer + reply reader (see WriterConfig::band / ReaderConfig::band for the dedicated-port / deferred-dispatch semantics; note each banded endpoint counts against Config::max_prioritized_endpoint_ports).

std::optional<espp::Dscp> dscp = {}

Optional DSCP code point applied to both endpoints’ dedicated sockets (marks the requests/replies this side SENDS).

class ServiceResponder

Handle to reply to a service request later (deferred reply). Copyable and movable; safe to store and fulfill from any thread. reply() sends the correlated response exactly once (subsequent calls are ignored). Used when the response is not ready when the request arrives - e.g. an action’s get_result, which must wait for the goal to finish. See add_service_server_deferred().

Public Functions

ServiceResponder() = default

Empty/invalid responder.

void reply(std::span<const uint8_t> response) const

Send the CDR-encapsulated response, correlated to the original request. No-op if invalid or already replied.

struct WriterConfig

Configuration for a writer (publishing endpoint).

Public Members

std::string topic

DDS topic name (e.g. “rt/chatter” for ROS 2).

std::string type_name

DDS type name (e.g. “std_msgs::msg::dds_::String_”).

Reliability reliability = {Reliability::BEST_EFFORT}

Reliability QoS.

uint16_t fragment_size = {63000}

Nominal per-fragment payload size (bytes) used when a published sample is too large for a single DATA submessage and is split into DATA_FRAG submessages. Default 63000 (large: fewer fragments). Lower it toward the path MTU (e.g. ~1400) for lossy links. Only relevant when fragmentation is compiled in (always on host; opt-in on ESP32). Ignored for samples that fit a single DATA submessage.

espp::QosBand band = {espp::QosBand::Normal}

Priority band for this writer’s endpoint (see espp::QosBand). Two distinct effects:

• Outbound pool scheduling (ALWAYS applies, no dedicated port needed): the writer’s progress()/send work is submitted to the transport pool at this band, so a higher-band writer’s outgoing DATA is scheduled ahead of lower-band work under load.

• Inbound socket dispatch (requires a DEDICATED unicast port): a non-Normal band (or a set dscp) requests a dedicated port so inbound protocol traffic addressed to it (ACKNACKs from reliable readers) is dispatched at this band and outgoing DATA leaves from that socket. Rationed - see Config::max_prioritized_endpoint_ports; when no dedicated port is available only the inbound-dispatch part is lost (ACKNACKs share the participant’s user-unicast port), while the outbound pool scheduling above still applies.

std::optional<espp::Dscp> dscp = {}

Optional DSCP code point (e.g. espp::Dscp::Ef) marking the traffic this writer SENDS. Requires (and by itself requests) a dedicated port, since DSCP is per-socket; ignored when none could be allocated.

class ServiceResponder

Handle to reply to a service request later (deferred reply). Copyable and movable; safe to store and fulfill from any thread. reply() sends the correlated response exactly once (subsequent calls are ignored). Used when the response is not ready when the request arrives - e.g. an action’s get_result, which must wait for the goal to finish. See add_service_server_deferred().

Public Functions

ServiceResponder() = default

Empty/invalid responder.

void reply(std::span<const uint8_t> response) const

Send the CDR-encapsulated response, correlated to the original request. No-op if invalid or already replied.

class ServiceClient

Client handle for calling a service. Obtain one from add_service_client(); it stays valid until the participant is stopped/destroyed.

Public Types

using reply_callback_t = std::function<void(std::span<const uint8_t> reply)>

Callback delivering a CDR-encapsulated reply for a call_async() request.

Public Functions

bool call_async(std::span<const uint8_t> request, reply_callback_t on_reply)

Send a request and invoke on_reply when the correlated reply arrives.

Returns:

False if the participant is not started or the request could not be queued. The callback runs on an engine worker thread.

std::optional<std::vector<uint8_t>> call(std::span<const uint8_t> request, std::chrono::milliseconds timeout)

Send a request and block until the correlated reply arrives or timeout.

Returns:

The CDR-encapsulated reply, or std::nullopt on timeout/failure. Do not call from within an engine callback (it would deadlock).

std::future<std::optional<std::vector<uint8_t>>> call_future(std::span<const uint8_t> request)

Send a request and return a future that becomes ready with the correlated reply (or std::nullopt if the request could not be queued). The future never blocks a worker thread; wait on it (or wait_for a timeout) from the caller. A pending request without a reply leaves the future unfulfilled until the participant stops.

class ActionGoalHandle

Server-side handle to a running goal, passed to the execute callback (which runs on its own thread). Publish feedback and terminate the goal through it.

Public Functions

std::span<const uint8_t> goal() const

The CDR-encapsulated goal payload.

void publish_feedback(std::span<const uint8_t> feedback) const

Publish a CDR-encapsulated feedback message for this goal.

void succeed(std::span<const uint8_t> result) const

Terminate the goal SUCCEEDED/ABORTED/CANCELED with a CDR result payload.

bool is_canceling() const

True if a cancel has been requested for this goal.

class ActionClient

Client handle for calling an action. Obtain from add_action_client().

Public Types

using feedback_callback_t = std::function<void(std::span<const uint8_t> feedback)>

CDR-encapsulated feedback for an in-progress goal.

using result_callback_t = std::function<void(int8_t status, std::span<const uint8_t> result)>

Terminal result: the GoalStatus value + the CDR-encapsulated result.

Public Functions

std::optional<GoalId> send_goal(std::span<const uint8_t> goal, feedback_callback_t on_feedback, result_callback_t on_result)

Send a goal. on_feedback is invoked for each feedback message; on_result once when the goal terminates (or is rejected, with an empty result).

Returns:

The generated goal id, or std::nullopt on failure.

bool cancel_goal(const GoalId &goal_id)

Request cancellation of a previously sent goal.

class NativeServiceClient

Client handle for a native service (see add_native_service_client()).

Public Functions

bool call_async(std::span<const uint8_t> request, reply_callback_t on_reply)

Send a request; invoke on_reply when the correlated reply arrives.

std::optional<std::vector<uint8_t>> call(std::span<const uint8_t> request, std::chrono::milliseconds timeout)

Send a request and block for the reply (std::nullopt on timeout/failure).

std::future<std::optional<std::vector<uint8_t>>> call_future(std::span<const uint8_t> request)

Send a request and return a future for the reply.

class NativeGoalHandle

Server-side handle to a running native goal (passed to the execute callback, which runs on its own thread).

Public Functions

void canceled(std::span<const uint8_t> result) const

Terminate the goal CANCELED (in response to a cancel request).

bool is_canceling() const

True if the client has requested cancellation of this goal (and the server’s on_cancel, if any, accepted it). A long-running execute callback should poll this and wind down - calling canceled()/abort() - when set.

class NativeActionClient

Client handle for a native action.

Public Types

using result_callback_t = std::function<void(uint8_t status, std::span<const uint8_t> result)>

Terminal result: the NativeGoalStatus value + the CDR result payload.

using accepted_callback_t = std::function<void(uint32_t goal_handle)>

Invoked once when the server accepts the goal, with the server-assigned goal_handle - keep it to cancel_goal() the goal later.

Public Functions

bool send_goal(std::span<const uint8_t> goal, feedback_callback_t on_feedback, result_callback_t on_result, accepted_callback_t on_accepted = nullptr)

Send a goal; on_feedback per feedback message, on_result once at the end, on_accepted (optional) with the assigned goal_handle when accepted.

Returns:

True if the goal was queued.

bool cancel_goal(uint32_t goal_handle)

Request cancellation of a previously accepted goal by its goal_handle (from on_accepted).

Returns:

True if the cancel request was queued.

Header File

• components/rtps/include/rtps_pubsub.hpp

Classes

template<RtpsMessage T>
class Publisher

Typed publisher: publish reflectable message structs on a topic.

A thin, header-only wrapper over espp::RtpsParticipant that removes the manual CDR (de)serialization + byte-span handling of the untyped API. Serialization uses the reflection-driven `cdr` component in ROS 2 / classic-CDR (XCDR1) wire format, into a reused buffer so steady-state publishing does not allocate.

struct Imu { float ax, ay, az; };   // any reflectable struct
espp::Publisher<Imu> pub(participant, {.topic = "rt/imu",
                                       .type_name = "sensor_msgs::msg::dds_::Imu_",
                                       .reliability = Reliability::RELIABLE});
pub.publish(Imu{0.1f, 0.2f, 9.8f});

Publisher Example

  // Typed reliable publisher: publish StringMsg structs directly (HEARTBEAT/
  // ACKNACK-acknowledged, retransmitted to matched readers). No manual CDR.
  using Reliability = espp::RtpsParticipant::Reliability;
  espp::Publisher<StringMsg> publisher(participant, {
                                                        .topic = pub_topic,
                                                        .type_name = type_name,
                                                        .reliability = Reliability::RELIABLE,
                                                    });
  // Typed subscriber: receive StringMsg structs directly.
  espp::Subscriber<StringMsg> subscriber(
      participant, {
                       .topic = sub_topic,
                       .type_name = type_name,
                       .on_message = [&](const StringMsg &msg) { logger.info("rx: {}", msg.data); },
                   });
  if (!publisher.is_valid() || !subscriber.is_valid()) {
    logger.error("Failed to create the typed publisher/subscriber");
    return;
  }

Note

For ROS 2 interop use ROS 2 naming: topic “rt/<name>” and type “<pkg>::msg::dds_::<Type>_” (e.g. “rt/chatter” + “std_msgs::msg::dds_::String_”).

Public Functions

inline Publisher(RtpsParticipant &participant, const Config &config)

Construct and register a writer on the participant. The participant must already be started and must outlive this publisher. Check is_valid() (or the return of publish()) to detect registration failure.

Parameters:

• participant – The started participant to publish through.

• config – The publisher configuration.

inline bool is_valid() const

Returns:

True if the writer was registered successfully.

inline bool publish(const T &sample)

Publish one sample. Serializes into a reused buffer (no steady-state allocation) and hands the CDR bytes to the participant. Thread-safe: concurrent calls are serialized (the reused buffer is mutex-guarded).

Parameters:

sample – The message to publish.

Returns:

True on success; false if invalid, serialization failed, the writer history was full, or the serialized size exceeds RtpsParticipant::max_payload_size.

struct Config

Configuration for a typed publisher.

Public Members

std::string topic

DDS topic name.

std::string type_name

DDS type name (must match the peer for interop).

RtpsParticipant::Reliability reliability = {RtpsParticipant::Reliability::BEST_EFFORT}

Reliability QoS.

espp::QosBand band = {espp::QosBand::Normal}

Priority band for the underlying writer endpoint (see RtpsParticipant::WriterConfig::band).

std::optional<espp::Dscp> dscp = {}

Optional DSCP marking for the traffic this publisher sends (see RtpsParticipant::WriterConfig::dscp).

template<RtpsMessage T>
class Subscriber

Typed subscriber: receive reflectable message structs from a topic.

A thin, header-only wrapper over espp::RtpsParticipant that deserializes each CDR sample into a T and delivers it to a typed callback, removing the manual byte-span + cdr::deserialize handling of the untyped API.

espp::Subscriber<Imu> sub(participant, {.topic = "rt/imu",
                                        .type_name = "sensor_msgs::msg::dds_::Imu_",
                                        .on_message = [](const Imu &m) { use(m); }});

Subscriber Example

  // Typed reliable publisher: publish StringMsg structs directly (HEARTBEAT/
  // ACKNACK-acknowledged, retransmitted to matched readers). No manual CDR.
  using Reliability = espp::RtpsParticipant::Reliability;
  espp::Publisher<StringMsg> publisher(participant, {
                                                        .topic = pub_topic,
                                                        .type_name = type_name,
                                                        .reliability = Reliability::RELIABLE,
                                                    });
  // Typed subscriber: receive StringMsg structs directly.
  espp::Subscriber<StringMsg> subscriber(
      participant, {
                       .topic = sub_topic,
                       .type_name = type_name,
                       .on_message = [&](const StringMsg &msg) { logger.info("rx: {}", msg.data); },
                   });
  if (!publisher.is_valid() || !subscriber.is_valid()) {
    logger.error("Failed to create the typed publisher/subscriber");
    return;
  }

Public Types

using message_callback_t = std::function<void(const T&)>

Called for each successfully deserialized sample.

Note

Runs on an engine worker thread - return quickly, do not block.

Public Functions

inline Subscriber(RtpsParticipant &participant, const Config &config)

Construct and register a reader on the participant. The participant must already be started.

Note

The registered reader (and thus this subscriber’s callback) lives on the participant until the participant is stopped - there is no per-reader removal. The callback holds a shared copy of the user callback, so destroying this Subscriber object is safe (it will not dangle); however, whatever the user callback itself references must outlive the participant. Stop the participant before tearing down state the callback captures.

Parameters:

• participant – The started participant to subscribe through.

• config – The subscriber configuration.

inline bool is_valid() const

Returns:

True if the reader was registered successfully.

struct Config

Configuration for a typed subscriber.

Public Members

std::string topic

DDS topic name.

std::string type_name

DDS type name (must match the peer for interop).

RtpsParticipant::Reliability reliability = {RtpsParticipant::Reliability::BEST_EFFORT}

Reliability QoS.

message_callback_t on_message = {nullptr}

Typed sample callback.

espp::QosBand band = {espp::QosBand::Normal}

Priority band for the underlying reader endpoint: dedicated receive port when available, deferred banded dispatch otherwise (see RtpsParticipant::ReaderConfig::band).

std::optional<espp::Dscp> dscp = {}

Optional DSCP marking for the traffic this subscriber sends (its ACKNACKs; see RtpsParticipant::ReaderConfig::dscp).

Header File

• components/rtps/include/rtps_service.hpp

Classes

template<RtpsMessage Request, RtpsMessage Response>
class ServiceServer

Typed service server (RMI): answers Request messages with Response messages, with no manual CDR handling.

A thin, header-only wrapper over espp::RtpsParticipant that (de)serializes the reflectable Request/Response structs around the byte-level service API. Works for both the ROS 2-interoperable and the native protocol (see Config::protocol).

struct AddReq { int64_t a, b; };
struct AddResp { int64_t sum; };
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}; }});

ServiceServer Example

  espp::ServiceServer<AddReq, AddResp> add_service(
      participant, {
                       .service = "/add_two_ints",
                       .type_name = "example_interfaces::srv::dds_::AddTwoInts",
                       .handler =
                           [&](const AddReq &r) {
                             logger.info("service add_two_ints: {} + {} = {}", r.a, r.b, r.a + r.b);
                             return AddResp{r.a + r.b};
                           },
                   });

Template Parameters:

• Request – Reflectable request message type (the service Request).

• Response – Reflectable response message type (the service Response).

Public Types

using handler_t = std::function<Response(const Request&)>

Handler: given a typed request, return the typed response. Runs on an engine worker thread - return promptly.

Public Functions

inline ServiceServer(RtpsParticipant &participant, const Config &config)

Construct and register the server on a started participant, which must outlive this object. Check is_valid().

Parameters:

• participant – The started participant to serve through.

• config – The server configuration (service name, type, handler).

inline bool is_valid() const

Returns:

True if the server registered successfully.

struct Config

Configuration for a typed service server.

Public Members

std::string service

Service name, e.g. “/add_two_ints”.

std::string type_name

Base DDS type (ROS 2), or any matching name (native).

handler_t handler

Request -> Response.

RtpsProtocol protocol = {RtpsProtocol::ROS2}

Wire protocol.

espp::QosBand band = {espp::QosBand::Normal}

Priority band for the server’s endpoints (see RtpsParticipant::ServiceConfig::band).

std::optional<espp::Dscp> dscp = {}

Optional DSCP marking for the replies this server sends (see RtpsParticipant::ServiceConfig::dscp).

template<RtpsMessage Request, RtpsMessage Response>
class ServiceClient

Typed service client (RMI): call a service with a Request and get a Response, with no manual CDR handling. Blocking, callback, and future styles.

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}, 1s)) use(resp->sum);

ServiceClient Example

  espp::ServiceClient<AddReq, AddResp> add_client(
      participant,
      {.service = "/peer_add_two_ints", .type_name = "example_interfaces::srv::dds_::AddTwoInts"});
  espp::ActionClient<FibGoal, FibSeq, FibSeq> fib_client(
      participant,
      {.action = "/peer_fib", .type_name = "example_interfaces::action::dds_::Fibonacci"});

  // Only one action goal in flight at a time: without a peer the goal never
  // completes, so re-sending on every tick would leak a pending goal each time.
  // The service call() below self-cleans on its 1s timeout, so it can run freely.
  std::atomic<bool> fib_in_flight{false};
  espp::Timer rpc_client_timer({
      .name = "rtps_rpc_client",
      .period = 5s,
      .callback =
          [&]() {
            // Typed blocking service call (RMI).
            if (auto resp = add_client.call(AddReq{20, 22}, 1s)) {
              logger.info("[client] /peer_add_two_ints(20,22) = {}", resp->sum);
            } else {
              logger.info("[client] /peer_add_two_ints: no reply (peer serving it?)");
            }
            // Typed action goal (AMI) with typed feedback + result. Skip if the
            // previous goal has not finished (e.g. no peer is serving it).
            if (!fib_in_flight.exchange(true)) {
              fib_client.send_goal(
                  FibGoal{5}, [&](const FibSeq &) { /* per-feedback */ },
                  [&](espp::GoalStatus status, const FibSeq &res) {
                    logger.info("[client] /peer_fib result: status={} len={}",
                                static_cast<int>(status), res.sequence.size());
                    fib_in_flight.store(false);
                  });
            }
            return false; // keep the timer running
          },
      .log_level = espp::Logger::Verbosity::WARN,
  });
  if (!add_client.is_valid() || !fib_client.is_valid()) {
    logger.error("Failed to create the typed service/action clients");
    return;
  }

Template Parameters:

• Request – Reflectable request message type (the service Request).

• Response – Reflectable response message type (the service Response).

Public Types

using response_callback_t = std::function<void(const Response&)>

Callback delivering the typed response for a call_async() request. Runs on an engine worker thread - return promptly.

Public Functions

inline ServiceClient(RtpsParticipant &participant, const Config &config)

Construct and register the client on a started participant, which must outlive this object. Check is_valid().

Parameters:

• participant – The started participant to call through.

• config – The client configuration (service name, type, protocol).

inline bool is_valid() const

Returns:

True if the client registered successfully.

inline std::optional<Response> call(const Request &request, std::chrono::milliseconds timeout)

Blocking call (RMI): send the request and wait for the correlated reply.

Parameters:

• request – The typed request.

• timeout – How long to wait for the reply.

Returns:

The Response, or std::nullopt on timeout / failure. Do not call from within an engine callback (it would deadlock).

inline bool call_async(const Request &request, const response_callback_t &on_response)

Async call (AMI): on_response(Response) is invoked when the correlated reply arrives (on an engine worker thread).

Parameters:

• request – The typed request.

• on_response – Called once with the typed response.

Returns:

False if the request could not be queued.

inline std::future<std::optional<Response>> call_future(const Request &request)

Future-based call (AMI): the future becomes ready with the Response (std::nullopt if the request could not be queued). Works for both the ROS 2 and native protocols (built on call_async). Wait on the future - or wait_for a timeout - from the caller; do not block a worker thread.

struct Config

Configuration for a typed service client.

Public Members

std::string service

Service name, e.g. “/add_two_ints”.

std::string type_name

Base DDS type (ROS 2), or any matching name (native).

RtpsProtocol protocol = {RtpsProtocol::ROS2}

Wire protocol.

espp::QosBand band = {espp::QosBand::Normal}

Priority band for the client’s endpoints (see RtpsParticipant::ServiceConfig::band).

std::optional<espp::Dscp> dscp = {}

Optional DSCP marking for the requests this client sends (see RtpsParticipant::ServiceConfig::dscp).

Header File

• components/rtps/include/rtps_action.hpp

Classes

template<RtpsMessage Goal, RtpsMessage Result, RtpsMessage Feedback>
class ActionGoalHandle

Server-side handle to a running typed goal, passed to the execute callback (which runs on its own thread). Publish feedback and terminate the goal through it, all with typed messages (no manual CDR). Exactly one terminator (succeed / abort) should be called per goal.

ActionGoalHandle Example

The handle is the `h` passed to the ActionServer execute callback:

  // Typed action (AMI) server: a ROS 2 client can `ros2 action send_goal
  // /fibonacci example_interfaces/action/Fibonacci "{order: 5}"` and receive
  // feedback + the [0,1,1,2,3,5] result. execute() runs on its own thread.
  espp::ActionServer<FibGoal, FibSeq, FibSeq> fib_action(
      participant, {
                       .action = "/fibonacci",
                       .type_name = "example_interfaces::action::dds_::Fibonacci",
                       .on_goal = [&](const FibGoal &g) { return g.order > 0; },
                       .execute =
                           [&](auto &h) {
                             const int32_t order = h.goal().order;
                             std::vector<int32_t> seq{0, 1};
                             for (int32_t i = 1; i < order; ++i) {
                               seq.push_back(seq[i] + seq[i - 1]);
                               h.publish_feedback(FibSeq{seq});
                               std::this_thread::sleep_for(200ms);
                             }
                             h.succeed(FibSeq{seq});
                             logger.info("action fibonacci({}) done", order);
                           },
                   });

Template Parameters:

• Goal – Reflectable goal message type.

• Result – Reflectable result message type.

• Feedback – Reflectable feedback message type.

Public Functions

inline const Goal &goal() const

Returns:

The typed goal being executed.

inline bool is_canceling() const

Returns:

True if the client has requested cancellation of this goal (both the ROS 2 and native protocols). A long-running execute callback should poll this and wind the goal down - calling canceled() - when it becomes true.

inline void publish_feedback(const Feedback &feedback) const

Publish a typed feedback message for this goal.

Parameters:

feedback – The feedback to send to the client.

inline void succeed(const Result &result) const

Terminate the goal as SUCCEEDED and deliver the result to the client.

Parameters:

result – The final result.

inline void abort(const Result &result) const

Terminate the goal as ABORTED and deliver the result to the client.

Parameters:

result – The (partial/error) result.

inline void canceled(const Result &result) const

Terminate the goal as CANCELED (in response to is_canceling()) and deliver the (partial) result to the client.

Parameters:

result – The result gathered before cancellation.

template<RtpsMessage Goal, RtpsMessage Result, RtpsMessage Feedback>
class ActionServer

Typed action server (AMI): runs long goals with typed Goal / Result / Feedback messages, no manual CDR handling.

struct Goal { int32_t order; };
struct Seq  { std::vector<int32_t> sequence; };  // Result + Feedback
espp::ActionServer<Goal, Seq, Seq> server(participant, {
    .action = "/fibonacci",
    .type_name = "example_interfaces::action::dds_::Fibonacci",
    .on_goal = [](const Goal &g) { return g.order > 0; },
    .execute = [](auto &h) {
       h.publish_feedback(...); h.succeed(...); }});

ActionServer Example

  // Typed action (AMI) server: a ROS 2 client can `ros2 action send_goal
  // /fibonacci example_interfaces/action/Fibonacci "{order: 5}"` and receive
  // feedback + the [0,1,1,2,3,5] result. execute() runs on its own thread.
  espp::ActionServer<FibGoal, FibSeq, FibSeq> fib_action(
      participant, {
                       .action = "/fibonacci",
                       .type_name = "example_interfaces::action::dds_::Fibonacci",
                       .on_goal = [&](const FibGoal &g) { return g.order > 0; },
                       .execute =
                           [&](auto &h) {
                             const int32_t order = h.goal().order;
                             std::vector<int32_t> seq{0, 1};
                             for (int32_t i = 1; i < order; ++i) {
                               seq.push_back(seq[i] + seq[i - 1]);
                               h.publish_feedback(FibSeq{seq});
                               std::this_thread::sleep_for(200ms);
                             }
                             h.succeed(FibSeq{seq});
                             logger.info("action fibonacci({}) done", order);
                           },
                   });

Template Parameters:

• Goal – Reflectable goal message type.

• Result – Reflectable result message type.

• Feedback – Reflectable feedback message type.

Public Types

using Handle = ActionGoalHandle<Goal, Result, Feedback>

The typed goal handle passed to the execute callback.

using goal_callback_t = std::function<bool(const Goal&)>

Called when a goal arrives; return true to accept it, false to reject. Runs on an engine worker thread - return promptly.

using execute_callback_t = std::function<void(Handle&)>

Called (on its own thread) to run an accepted goal to completion via the handle (publish_feedback / succeed / abort).

Public Functions

inline ActionServer(RtpsParticipant &participant, const Config &config)

Construct and register the action server on a started participant, which must outlive this object. Check is_valid().

Parameters:

• participant – The started participant to serve through.

• config – The server configuration.

inline bool is_valid() const

Returns:

True if the action server registered successfully.

struct Config

Configuration for a typed action server.

Public Members

std::string action

Action name, e.g. “/fibonacci”.

std::string type_name

Base DDS type (ROS 2), or any matching name (native).

goal_callback_t on_goal

Accept/reject each incoming goal.

execute_callback_t execute

Run each accepted goal (own thread).

RtpsProtocol protocol = {RtpsProtocol::ROS2}

Wire protocol.

espp::QosBand band = {espp::QosBand::Normal}

Priority band inherited by all of the action’s underlying endpoints (see RtpsParticipant::ActionConfig::band, incl. the ration note).

std::optional<espp::Dscp> dscp = {}

Optional DSCP marking for the traffic the server sends (see RtpsParticipant::ActionConfig::dscp).

template<RtpsMessage Goal, RtpsMessage Result, RtpsMessage Feedback>
class ActionClient

Typed action client (AMI): send typed goals and receive typed feedback + result, no manual CDR handling.

espp::ActionClient<Goal, Seq, Seq> client(participant, {
    .action = "/fibonacci",
    .type_name = "example_interfaces::action::dds_::Fibonacci"});
client.send_goal(Goal{5},
    [](const Seq &fb) { ... },
    [](espp::GoalStatus st, const Seq &res) { ... });

ActionClient Example

  espp::ServiceClient<AddReq, AddResp> add_client(
      participant,
      {.service = "/peer_add_two_ints", .type_name = "example_interfaces::srv::dds_::AddTwoInts"});
  espp::ActionClient<FibGoal, FibSeq, FibSeq> fib_client(
      participant,
      {.action = "/peer_fib", .type_name = "example_interfaces::action::dds_::Fibonacci"});

  // Only one action goal in flight at a time: without a peer the goal never
  // completes, so re-sending on every tick would leak a pending goal each time.
  // The service call() below self-cleans on its 1s timeout, so it can run freely.
  std::atomic<bool> fib_in_flight{false};
  espp::Timer rpc_client_timer({
      .name = "rtps_rpc_client",
      .period = 5s,
      .callback =
          [&]() {
            // Typed blocking service call (RMI).
            if (auto resp = add_client.call(AddReq{20, 22}, 1s)) {
              logger.info("[client] /peer_add_two_ints(20,22) = {}", resp->sum);
            } else {
              logger.info("[client] /peer_add_two_ints: no reply (peer serving it?)");
            }
            // Typed action goal (AMI) with typed feedback + result. Skip if the
            // previous goal has not finished (e.g. no peer is serving it).
            if (!fib_in_flight.exchange(true)) {
              fib_client.send_goal(
                  FibGoal{5}, [&](const FibSeq &) { /* per-feedback */ },
                  [&](espp::GoalStatus status, const FibSeq &res) {
                    logger.info("[client] /peer_fib result: status={} len={}",
                                static_cast<int>(status), res.sequence.size());
                    fib_in_flight.store(false);
                  });
            }
            return false; // keep the timer running
          },
      .log_level = espp::Logger::Verbosity::WARN,
  });
  if (!add_client.is_valid() || !fib_client.is_valid()) {
    logger.error("Failed to create the typed service/action clients");
    return;
  }

Template Parameters:

• Goal – Reflectable goal message type.

• Result – Reflectable result message type.

• Feedback – Reflectable feedback message type.

Public Types

using feedback_callback_t = std::function<void(const Feedback&)>

Callback delivering one typed feedback message (on an engine worker thread).

using result_callback_t = std::function<void(GoalStatus, const Result&)>

Callback delivering the terminal status + typed result, once per goal.

Public Functions

inline ActionClient(RtpsParticipant &participant, const Config &config)

Construct and register the action client on a started participant, which must outlive this object. Check is_valid().

Parameters:

• participant – The started participant to drive the action through.

• config – The client configuration.

inline bool is_valid() const

Returns:

True if the action client registered successfully.

inline bool send_goal(const Goal &goal, feedback_callback_t on_feedback, result_callback_t on_result)

Send a typed goal to the server.

Parameters:

• goal – The typed goal.

• on_feedback – Invoked for each feedback message during execution.

• on_result – Invoked once with the terminal status + result (an empty Result and non-SUCCEEDED status if the goal was rejected).

Returns:

True if the goal was queued.

inline bool cancel_goal()

Request cancellation of the most recently accepted goal (works on both the ROS 2 and native protocols). The server observes the cancel via its goal handle’s is_canceling() and should wind the goal down cooperatively.

Returns:

True if the cancel request was queued.

struct Config

Configuration for a typed action client.

Public Members

std::string action

Action name, e.g. “/fibonacci”.

std::string type_name

Base DDS type (ROS 2), or any matching name (native).

RtpsProtocol protocol = {RtpsProtocol::ROS2}

Wire protocol.

espp::QosBand band = {espp::QosBand::Normal}

Priority band inherited by all of the action’s underlying endpoints (see RtpsParticipant::ActionConfig::band, incl. the ration note).

std::optional<espp::Dscp> dscp = {}

Optional DSCP marking for the traffic the client sends (see RtpsParticipant::ActionConfig::dscp).

Header File

• components/rtps/include/rtps_message.hpp