feat(rtps_embedded): Typed Publisher<T>/Subscriber<T> pub/sub (Phase 5)

[finger563]

Phase 5 (typed pub/sub) of the components/rtps_embedded refactor — see REFACTOR_PLAN.md. Independent of and parallel to the Phase 4 (limits) work.

What

A header-only typed layer over espp::RtpsParticipant so applications publish/receive reflectable message structs directly — no manual CDR (de)serialization or byte-span handling.

struct Imu { float ax, ay, az; };   // any reflectable struct — no base class/macros

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});

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

• RtpsMessage concept: any struct the reflection cdr can (de)serialize qualifies.
• Publisher<T> serializes with cdr::serialize_into into a reused buffer → zero steady-state allocation.
• espp::ros2::topic_name() helper for ROS 2 naming.

Additive by design

The untyped span/byte API (add_writer/add_reader/publish/on_sample) is unchanged and is exactly what the typed layer wraps. Existing code is unaffected; the golden wire tests are byte-identical.

The esp32 example is converted to the typed API (cleaner than the manual-cdr::serialize version, and it verifies the header compiles with reflect-cpp on target). Python keeps the byte API — templates don't bind generically, so the typed layer is a C++ ergonomic layer.

Verification

• pc/tests/rtps_typed_pubsub: two participants exchange a multi-field struct (uint32 + float + string) end-to-end over real SEDP/DATA, with field-level verification of the typed round-trip — PASS.
• Golden wire tests byte-identical.
• esp32 example builds for esp32p4 on ESP-IDF 6.0; host lib builds.

🤖 Generated with Claude Code

[github-actions]

✅Static analysis result - no issues found! ✅

[Copilot]

Pull request overview

Adds a header-only typed pub/sub facade (espp::Publisher<T> / espp::Subscriber<T>) on top of espp::RtpsParticipant in components/rtps_embedded, enabling applications to publish/receive reflectable structs directly (no manual CDR byte handling), plus host-side and ESP32 example validation.

Changes:

• Introduces components/rtps_embedded/include/rtps_pubsub.hpp with RtpsMessage concept and typed Publisher<T> / Subscriber<T> wrappers.
• Updates the rtps_embedded ESP32 example to use the typed API instead of manual CDR serialization.
• Adds a new host PC test pc/tests/rtps_typed_pubsub.cpp and exposes the new header via lib/include/espp.hpp.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

File	Description
pc/tests/rtps_typed_pubsub.cpp	New host-side end-to-end typed pub/sub test validating round-trip of a multi-field struct.
lib/include/espp.hpp	Adds rtps_pubsub.hpp to the umbrella include set.
components/rtps_embedded/include/rtps_pubsub.hpp	New typed pub/sub layer (concept + Publisher/Subscriber + ROS 2 topic helper).
components/rtps_embedded/example/main/main.cpp	Migrates the ESP32 example from manual CDR byte handling to typed pub/sub.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[finger563]

Addressed all review comments + the static_analysis failure in 833bb2d:

• Missing standard headers (Copilot): added <concepts>/<cstddef>/<memory>/<mutex> — the concept (std::convertible_to) and std::byte/std::size_t were only reachable transitively.
• Subscriber use-after-free (Copilot): the on_sample callback now captures a shared_ptr copy of the user callback instead of this, so destroying a Subscriber while the participant is still running can't dangle the engine-invoked callback. Documented the lifetime contract (stop the participant before tearing down state the callback references).
• Publisher data race (Copilot): publish() now guards the reused serialization buffer with a mutex, so concurrent calls from multiple threads are safe.
• static_analysis (cppcheck): the findings were in the test — Telemetry members now default-initialized (uninitMemberVarNoCtor), and the verification loop uses std::any_of (useStlAlgorithm).

Verified: cppcheck clean on both files with the CI args; typed test passes (sent=5 received=5 with field-level verification); golden byte-identical.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.

Suppressed comments (3)

components/rtps_embedded/include/rtps_pubsub.hpp:28

• The RtpsMessage concept doesn’t currently constrain the parts of the cdr API that Publisher<T> actually uses (cdr::serialize_into<cdr::xcdr1>(sample, buffer_)) nor does it constrain the deserialize return type. Tightening the concept to require serialize_into (and to require that deserialize returns an optional/expected-like type holding T) will produce clearer compile-time errors and prevent types that only partially satisfy the typed pub/sub contract from compiling until publish() is instantiated.

template <typename T>
concept RtpsMessage = requires(const T &value, std::span<const std::byte> bytes) {
  { cdr::serialized_size<cdr::xcdr1>(value) } -> std::convertible_to<std::size_t>;
  {cdr::deserialize<T>(bytes)};
};

components/rtps_embedded/include/rtps_pubsub.hpp:185

• As written, topic_name() will generate rt//foo if callers pass a canonical ROS topic like /foo, and will double-prefix if callers pass an already-mangled DDS name like rt/foo. If this helper is meant to be turnkey, consider normalizing input by stripping a leading / and returning the input unchanged when it already starts with rt/.

inline std::string topic_name(std::string_view ros_topic) {
  std::string out = "rt/";
  out += ros_topic;
  return out;
}

pc/tests/rtps_typed_pubsub.cpp:72

• A fixed sleep_for(2s) can make this test flaky (slow CI machines, loaded hosts, or different network stacks may need more/less than 2 seconds). Prefer folding discovery/wait time into the existing deadline loop (e.g., start publishing immediately and rely on reliable delivery once matched, or poll/wait on an explicit 'matched' signal if the participant exposes one) so the test outcome depends on kDeadline rather than a hard-coded pre-delay.

  // Let SEDP match before publishing.
  std::this_thread::sleep_for(2s);

[finger563]

Addressed the second review round + added the Python typed layer (f75157a):

Review comments (round 2):

• strict-aliasing reinterpret_cast (Copilot, both directions): replaced with the standard, well-defined span conversions — std::as_writable_bytes on the publish path (buffer_ is now std::vector<uint8_t>) and std::as_bytes on the receive path. No reinterpret_cast remains.
• example static lifetime (Copilot): participant/publisher/subscriber are now automatic locals, so they RAII-clean up in the correct order on any early return.

Python typed pub/sub (answering the "what's the Python equivalent?" discussion): since Publisher<T>/Subscriber<T> are C++ templates and can't be bound generically, the Python counterpart is a small pure-Python wrapper — espp.rtps.Publisher / Subscriber — over the byte API paired with a runtime CDR codec. It's codec-agnostic (any type with .serialize()/.deserialize(), or explicit callables); pycdr2 dataclasses satisfy it and emit ROS 2 / DDS-compatible CDR. The python examples now publish/receive a complex nested message (sensor_msgs/msg/Imu-style: string + nested struct + fixed array + sequence), pycdr2 added to python/requirements.txt, and rtps_typed_test.py verifies the fields round-trip end-to-end.

Verified: cppcheck clean; C++ + Python typed tests pass (fields verified); golden byte-identical; esp32 example builds.

[Copilot]

Pull request overview

Copilot reviewed 11 out of 12 changed files in this pull request and generated 2 comments.

Suppressed comments (4)

lib/python_bindings/espp/rtps.py:128

• Unlike Publisher, this wrapper does not retain participant. If the caller releases its last participant reference while keeping the subscriber, the underlying participant is destroyed/stopped while sub.valid remains true. Store the participant on self so the subscription keeps its transport alive.

        deser = deserialize if deserialize is not None else (lambda data: message_type.deserialize(data))

components/rtps_embedded/include/rtps_pubsub.hpp:41

• The public example does not compile as written because Reliability is nested under RtpsParticipant and no alias is introduced in this snippet. Qualify the enum so users can copy the example directly.

///                                        .reliability = Reliability::RELIABLE});

components/rtps_embedded/include/rtps_pubsub.hpp:30

• This new public API is not listed in doc/Doxyfile's INPUT, so its Doxygen documentation will not be generated or exposed with the rest of the library API. Add components/rtps_embedded/include/rtps_pubsub.hpp in alphabetical order and update the component documentation for the typed layer.

/// @brief Typed publisher: publish reflectable message structs on a topic.

lib/python_bindings/espp/rtps.py:67

• The PR description says Python remains on the byte API and that the typed layer is C++-only, but this adds and publicly exports a Python typed Publisher/Subscriber API and converts the Python examples to it. Update the PR description so its stated scope and compatibility claims match the code being reviewed.

class Publisher:

[finger563]

Addressed the new comments in 68fbf9b:

• Payload > 64 KB silently truncated (Copilot / raised for the parallel limits work): RtpsParticipant::publish cast the length to DataSize_t (uint16_t), so oversized samples wrapped. Added RtpsParticipant::max_payload_size (65535, static_assert'd against the engine's DataSize_t) and now reject oversized payloads with a logged error in the untyped publish() (protects all callers); Publisher<T>::publish also early-rejects before serializing. This 64 KB cap is inherent to the RTPS wire format — the DATA submessage octetsToNextHeader is 16-bit — so raising it (for images/large data on hosts) is a real protocol feature (DATA_FRAG fragmentation + widening the internal size type), tracked for the limits/refactor track rather than this PR.
• Demo message reused the official sensor_msgs/msg/Imu type name with a mismatched layout (Copilot): real DDS/ROS 2 peers would match on the name and misdecode. Renamed to demo type names (espp_examples/msg/Vector3, espp_examples/msg/SensorSample) with a note not to reuse official ROS 2 type names unless the schema matches exactly.

The earlier aliasing / static / data-race / include / UAF comments were already resolved in 833bb2d and f75157a (Copilot re-posts them on the pre-fix diff positions). Verified: C++ + Python typed tests pass (fields verified), golden byte-identical, cppcheck clean, esp32 builds.