Skip to content

Feat/embedded rtps merge espp - #672

Draft
guo-max wants to merge 23 commits into
mainfrom
feat/embeddedRTPS_merge_espp
Draft

Feat/embedded rtps merge espp#672
guo-max wants to merge 23 commits into
mainfrom
feat/embeddedRTPS_merge_espp

Conversation

@guo-max

@guo-max guo-max commented Jul 21, 2026

Copy link
Copy Markdown
Collaborator

Description

integrates the embeddedRTPS RTPS/DDS stack into the ESPP ecosystem.

Motivation and Context

Try to have an embedded rtps that could be used to talk between embedded board as well as a PC/Jetson that running ROS2 or any other DDS participant on the same network using the standard RTPS wire protocal

How has this been tested?

Tested pub/sub between
ESP32 <-----> ESP32
ESP32 <-----> PC (embedded rtps)
ESP32 <-----> PC (fastRTPS)
ESP32 <-----> PC (ROS2)

Screenshots (if appropriate, e.g. schematic, board, console logs, lab pictures):

ROS:
image
ESP32:
image

Types of changes

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality)
  • Breaking change (fix or feature that would cause existing functionality to not work as expected)
  • Documentation Update
  • Hardware (schematic, board, system design) change
  • Software change

Checklist:

  • My change requires a change to the documentation.
  • I have added / updated the documentation related to this change via either README or WIKI

Software

  • I have added tests to cover my changes.
  • I have updated the .github/workflows/build.yml file to add my new test to the automated cloud build github action.
  • All new and existing tests passed.
  • My code follows the code style of this project.

Hardware

  • I have updated the design files (schematic, board, libraries).
  • I have attached the PDFs of the SCH / BRD to this PR
  • I have updated the design output (GERBER, BOM) files.

Copilot AI lite review requested due to automatic review settings July 21, 2026 19:53
@github-actions

github-actions Bot commented Jul 21, 2026

Copy link
Copy Markdown

⚡ Static analysis result ⚡

🔴 cppcheck found 43 issues! Click here to see details.

namespace rtps {
class Writer;
} // namespace rtps

!Line: 39 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
} // namespace rtps
#endif // RTPS_RTPS_H

!Line: 31 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
#define IS_LITTLE_ENDIAN 1
namespace Config {
const VendorId_t VENDOR_ID = {13, 37};

!Line: 31 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
#define IS_LITTLE_ENDIAN 1
#define OS_IS_FREERTOS
namespace Config {

!Line: 31 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
inline size_t hashCharArray(const char *p, size_t s) {
size_t result = 0;
const size_t prime = 31;
for (size_t i = 0; i < s; ++i) {
result = p[i] + (result * prime);

!Line: 31 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

printf("%x", static_cast<uint8_t>(id.entityKind));
printf("\n");
}
inline void printGuidPrefix(rtps::GuidPrefix_t prefix) {

!Line: 14 - error: Code 'C++ cast <...' is invalid C code. [syntaxError]

namespace rtps {
namespace Diagnostics {
namespace ThreadPool {
extern uint32_t dropped_incoming_packets_usertraffic;
extern uint32_t dropped_incoming_packets_metatraffic;

!Line: 31 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
inline Time_t getCurrentTimeStamp() {
static const auto start = std::chrono::steady_clock::now();
const auto elapsed = std::chrono::steady_clock::now() - start;
const auto nowMs =
std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();

!Line: 33 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
namespace {
const uint16_t PB = 7400; // Port Base Number
const uint16_t DG = 250; // DomainId Gain
const uint16_t PG = 2; // ParticipantId Gain
// Additional Offsets

!Line: 33 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
#if defined(_MSC_VER)
#define RTPS_EMBEDDED_PACKED
#pragma pack(push, 1)
#else

!Line: 35 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
// TODO move types to where they are needed!
typedef uint16_t Ip4Port_t;
typedef uint16_t DataSize_t;

!Line: 36 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Reader;
class Writer;
class Participant;
class MessageProcessingInfo;

!Line: 35 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
#if defined(_MSC_VER)
#define RTPS_EMBEDDED_PACKED
#pragma pack(push, 1)
#else

!Line: 34 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
namespace MessageFactory {
const std::array<uint8_t, 4> PROTOCOL_TYPE{'R', 'T', 'P', 'S'};
const uint8_t numBytesUntilEndOfLength =
4; // The first bytes incl. submessagelength don't count

!Line: 40 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
/**
* Simple version of a history cache. It sets consecutive sequence numbers
* automatically which allows an easy and fast approach of dropping acknowledged
* changes. Furthermore, disposing of arbitrary changes is not possible.

!Line: 32 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
template <typename T, uint16_t SIZE> class ThreadSafeCircularBuffer {
public:
bool init();

!Line: 34 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
template <class TYPE, uint32_t SIZE> class MemoryPool {
public:
template <typename IT_TYPE> class MemoryPoolIterator {
public:

!Line: 33 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct CacheChange {
using TimePoint = std::chrono::steady_clock::time_point;
ChangeKind_t kind = ChangeKind_t::INVALID;
bool inLineQoS = false;

!Line: 34 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct PayloadBuffer {
std::vector<uint8_t> bytes;
bool isValid() const { return true; }

!Line: 33 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
/**
* Extension of the SimpleHistoryCache that allows for deletion operation at the
* cost of efficieny
* TODO: Replace with something better in the future!

!Line: 32 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct ReaderProxy {
Guid_t remoteReaderGuid;
Count_t ackNackCount = {0};
LocatorIPv4 remoteLocator;
bool is_reliable = false;

!Line: 32 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class EsppTransport;
template <class NetworkDriver> class StatefulWriterT final : public Writer {
public:

!Line: 38 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Domain : public espp::BaseComponent {
public:
explicit Domain(const Ip4AddressBytes &localIpAddress);
Domain(EsppTransport &transport,
const Ip4AddressBytes &localIpAddress);

!Line: 42 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class EsppTransport;
struct SubmessageHeartbeat;
template <class NetworkDriver> class StatefulReaderT final : public Reader {
public:

!Line: 36 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Writer : public espp::BaseComponent {
public:
TopicData m_attributes;
virtual bool addNewMatchedReader(const ReaderProxy &newProxy);

!Line: 55 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Writer;
class Reader;
class Participant : public espp::BaseComponent {

!Line: 40 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct SubmessageHeartbeat;
struct SubmessageGap;
class ReaderCacheChange {

!Line: 38 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class EsppTransport;
template <typename NetworkDriver> class StatelessWriterT : public Writer {
public:

!Line: 36 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class StatelessReader final : public Reader {
public:
bool init(const TopicData &attributes);
void newChange(const ReaderCacheChange &cacheChange) override;
bool onNewHeartbeat(const SubmessageHeartbeat &msg,

!Line: 31 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct WriterProxy {
Guid_t remoteWriterGuid;
SequenceNumber_t expectedSN;
Count_t ackNackCount;
Count_t hbCount;

!Line: 32 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Participant;
using SMElement::ParameterId;
using BuiltinEndpointSet_t = uint32_t;

!Line: 39 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Participant;
class Writer;
class Reader;
class ReaderCacheChange;

!Line: 48 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct BuiltInTopicKey {
std::array<uint32_t, 3> value;
};

!Line: 37 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class Participant;
class ReaderCacheChange;
class Writer;
class Reader;

!Line: 37 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct BuiltInEndpoints {
Writer *spdpWriter = nullptr;
Reader *spdpReader = nullptr;
Writer *sedpPubWriter = nullptr;

!Line: 34 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
class EsppTransport : public espp::BaseComponent {
public:
using RxCallback = ReceiveCallback;

!Line: 41 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

namespace rtps {
struct PacketInfo {
Ip4Port_t srcPort; // TODO Do we need that?
std::array<uint8_t, 4> destAddr = {0, 0, 0, 0};
Ip4Port_t destPort;

!Line: 37 - error: Code 'namespacertps{' is invalid C code. [syntaxError]

SEDP_LOG("Added new change to sedpPubWriter.\n");
#endif
}
template <typename A>
bool SEDPAgent::disposeEndpointInSEDPHistory(A *local_endpoint,

!Line: 361 - style: inconclusive: Statements following 'return' will never be executed. [unreachableCode]

SEDP_LOG("Added new change to sedpSubWriter.\n");
#endif
}

!Line: 504 - style: inconclusive: Statements following 'return' will never be executed. [unreachableCode]

if (!channel.socket || !channel.socket->is_valid()) {
logger_.error("Failed to create valid UDP socket for port {}", receivePort);
channel.socket.reset();
return nullptr;
}

!Line: 140 - style: Condition '!channel.socket' is always false [knownConditionTrueFalse]

if (channel.in_use && channel.port == port) {
return &channel;
}
}
return nullptr;
}

!Line: 78 - style: Consider using std::find_if algorithm instead of a raw loop. [useStlAlgorithm]

if (channel.in_use && channel.port == port) {
return &channel;
}
}
return nullptr;
}

!Line: 87 - style: Consider using std::find_if algorithm instead of a raw loop. [useStlAlgorithm]

for (const auto &existing : m_multicastGroups) {
if (existing == group) {
return true;
}
}

!Line: 197 - style: Consider using std::any_of algorithm instead of a raw loop. [useStlAlgorithm]



rtps::Writer *Participant::getWriter(EntityId_t id) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_writers.size(); ++i) {

rtps::Reader *Participant::getReader(EntityId_t id) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_readers.size(); ++i) {

rtps::Reader *Participant::getReaderByWriterId(const Guid_t &guid) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_readers.size(); ++i) {

rtps::Writer *Participant::getMatchingWriter(const TopicData &readerTopicData) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_writers.size(); ++i) {

rtps::Reader *Participant::getMatchingReader(const TopicData &writerTopicData) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_readers.size(); ++i) {
rtps::Writer *
Participant::getMatchingWriter(const TopicDataCompressed &readerTopicData) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_writers.size(); ++i) {
rtps::Reader *
Participant::getMatchingReader(const TopicDataCompressed &writerTopicData) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_readers.size(); ++i) {
bool Participant::hasReaderWithMulticastLocator(
const std::array<uint8_t, 4> &address) {
std::lock_guard<std::recursive_mutex> lock(m_mutex);
for (uint8_t i = 0; i < m_readers.size(); i++) {

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Integrates the embeddedRTPS RTPS/DDS stack into ESPP as a new component (rtps_embedded), adds a new reusable thread_pool component, and introduces PC-side standalone publisher/subscriber tests to exercise the stack outside ESP-IDF.

Changes:

  • Add components/rtps_embedded with ESPP-based transport/task/threading adapters plus an ESP-IDF example app.
  • Add components/thread_pool as a reusable worker-queue abstraction with an ESP-IDF example.
  • Extend pc/ build to compile embeddedRTPS-based host tests and include the necessary sources (including Micro-CDR).

Reviewed changes

Copilot reviewed 82 out of 83 changed files in this pull request and generated 8 comments.

Show a summary per file
File Description
pc/tests/rtps_embedded_subscriber.cpp Adds standalone PC subscriber test for embeddedRTPS.
pc/tests/rtps_embedded_publisher.cpp Adds standalone PC publisher test for embeddedRTPS.
pc/CMakeLists.txt Adds embeddedRTPS + Micro-CDR sources to PC tests; enables C compilation for those tests.
components/thread_pool/src/thread_pool.cpp Implements espp::ThreadPool.
components/thread_pool/README.md Documents the new thread pool component.
components/thread_pool/include/thread_pool.hpp Public API for the new thread pool component.
components/thread_pool/idf_component.yml IDF Component Manager manifest for thread_pool.
components/thread_pool/example/sdkconfig.defaults Example sdkconfig enabling C++ exceptions.
components/thread_pool/example/README.md Build instructions for the thread_pool example.
components/thread_pool/example/main/thread_pool_example.cpp Example showing how to submit jobs and await completion.
components/thread_pool/example/main/CMakeLists.txt Registers example main component.
components/thread_pool/example/CMakeLists.txt ESP-IDF project CMake for the thread_pool example.
components/thread_pool/CMakeLists.txt Registers the thread_pool component sources/includes.
components/socket/CMakeLists.txt Adjusts socket component requirements list.
components/rtps_embedded/src/utils/Diagnostics.cpp Adds embeddedRTPS diagnostics storage definitions.
components/rtps_embedded/src/messages/MessageTypes.cpp Adds embeddedRTPS message (de)serialization implementation.
components/rtps_embedded/src/messages/MessageReceiver.cpp Adds RTPS submessage processing/dispatch.
components/rtps_embedded/src/entities/Writer.cpp Adds Writer proxy management and base Writer logic.
components/rtps_embedded/src/entities/StatelessReader.cpp Adds StatelessReader implementation.
components/rtps_embedded/src/entities/Reader.cpp Adds Reader callback/proxy management implementation.
components/rtps_embedded/src/discovery/ParticipantProxyData.cpp Adds SPDP participant proxy deserialization and locator parsing.
components/rtps_embedded/src/communication/EsppTransport.cpp Implements ESPP-backed UDP transport adapter for embeddedRTPS.
components/rtps_embedded/README.md High-level component documentation (architecture/config/quick-start).
components/rtps_embedded/include/rtps/utils/udpUtils.h Port/address helper utilities.
components/rtps_embedded/include/rtps/utils/sysFunctions.h Platform timestamp helper implementation.
components/rtps_embedded/include/rtps/utils/printutils.h Debug print helpers for RTPS IDs.
components/rtps_embedded/include/rtps/utils/Log.h Compile-time verbosity switches for embeddedRTPS logging.
components/rtps_embedded/include/rtps/utils/hash.h Hash utility used for compressed topic matching.
components/rtps_embedded/include/rtps/utils/Diagnostics.h Diagnostics extern declarations.
components/rtps_embedded/include/rtps/utils/constants.h Placeholder constants header.
components/rtps_embedded/include/rtps/ThreadPool.h embeddedRTPS threadpool facade integrating with espp::ThreadPool.
components/rtps_embedded/include/rtps/storages/ThreadSafeCircularBuffer.tpp Thread-safe circular buffer implementation.
components/rtps_embedded/include/rtps/storages/ThreadSafeCircularBuffer.h Thread-safe circular buffer API.
components/rtps_embedded/include/rtps/storages/SimpleHistoryCache.h Stateless history cache implementation.
components/rtps_embedded/include/rtps/storages/PayloadBuffer.h Payload buffer wrapper around std::vector<uint8_t>.
components/rtps_embedded/include/rtps/storages/MemoryPool.h Fixed-size memory pool container for embeddedRTPS entities.
components/rtps_embedded/include/rtps/storages/HistoryCacheWithDeletion.h Deletable history cache used by stateful endpoints.
components/rtps_embedded/include/rtps/storages/CacheChange.h CacheChange model for payload + sequencing.
components/rtps_embedded/include/rtps/rtps.h Convenience header for rtps::Domain.
components/rtps_embedded/include/rtps/messages/MessageReceiver.h MessageReceiver API.
components/rtps_embedded/include/rtps/messages/MessageFactory.h RTPS message construction helpers.
components/rtps_embedded/include/rtps/entities/WriterProxy.h WriterProxy model and missing-sequence computation.
components/rtps_embedded/include/rtps/entities/Writer.h Writer base class interface.
components/rtps_embedded/include/rtps/entities/StatelessWriter.tpp Stateless writer implementation template.
components/rtps_embedded/include/rtps/entities/StatelessWriter.h Stateless writer API.
components/rtps_embedded/include/rtps/entities/StatelessReader.h Stateless reader API.
components/rtps_embedded/include/rtps/entities/StatefulWriter.h Stateful writer API.
components/rtps_embedded/include/rtps/entities/StatefulReader.h Stateful reader API.
components/rtps_embedded/include/rtps/entities/ReaderProxy.h ReaderProxy model.
components/rtps_embedded/include/rtps/entities/Reader.h Reader base class interface + ReaderCacheChange definition.
components/rtps_embedded/include/rtps/entities/Participant.h Participant API including discovery agents and endpoint management.
components/rtps_embedded/include/rtps/entities/Domain.h Domain API for participants/readers/writers plus lifecycle methods.
components/rtps_embedded/include/rtps/discovery/TopicData.h Topic metadata structures and matching logic.
components/rtps_embedded/include/rtps/discovery/SPDPAgent.h SPDP discovery agent API.
components/rtps_embedded/include/rtps/discovery/SEDPAgent.h SEDP discovery agent API.
components/rtps_embedded/include/rtps/discovery/ParticipantProxyData.h ParticipantProxyData model + liveliness checks.
components/rtps_embedded/include/rtps/discovery/BuiltInEndpoints.h Built-in endpoint pointers used by discovery.
components/rtps_embedded/include/rtps/config.h Selects config header based on build target.
components/rtps_embedded/include/rtps/config_esp32.h ESP32-focused config constants.
components/rtps_embedded/include/rtps/config_desktop.h Desktop-focused config constants.
components/rtps_embedded/include/rtps/communication/PacketInfo.h PacketInfo structure for transport send/receive.
components/rtps_embedded/include/rtps/communication/EsppTransport.h Transport adapter API wrapping espp::UdpSocket.
components/rtps_embedded/include/rtps/common/Locator.h Locator types and helpers for IPv4 locators.
components/rtps_embedded/example/sdkconfig.defaults Example sdkconfig for rtps_embedded (partition table, stacks).
components/rtps_embedded/example/partitions.csv Custom partition table for the rtps_embedded example.
components/rtps_embedded/example/main/main.cpp ESP-IDF example app integrating WiFi + RTPS pub/sub demo.
components/rtps_embedded/example/main/Kconfig.projbuild Menuconfig options for role/topics/multicast/network settings.
components/rtps_embedded/example/main/CMakeLists.txt Registers example main component and dependencies.
components/rtps_embedded/example/CMakeLists.txt ESP-IDF project CMake for rtps_embedded example.
components/rtps_embedded/CMakeLists.txt Registers rtps_embedded component sources/includes and dependencies.
.gitmodules Adds Micro-CDR as a submodule under rtps_embedded thirdparty.
.gitignore Ignores generated Micro-CDR config header under the submodule.

Comment on lines +94 to +101
bool serializeIntoUdcrBuffer(ucdrBuffer &buffer) {
if (ucdr_buffer_remaining(&buffer) < sizeof(FullLengthLocator)) {
return false;
} else {
ucdr_serialize_array_uint8_t(&buffer, reinterpret_cast<uint8_t *>(this),
sizeof(FullLengthLocator));
}
}
Comment on lines +6 to +9
#include <chrono>
#include <cstdint>
#include <cstdlib>
#include <thread>
Comment on lines +52 to +55
// Keep topic/type names short because embeddedRTPS desktop config caps lengths.
rtps::Writer *writer =
domain.createWriter(*participant, topic.c_str(), "UInt32", false);
if (writer == nullptr) {
Comment on lines +68 to +77
uint32_t value = 0;
while (true) {
++value;
const rtps::CacheChange *change =
writer->newChange(rtps::ChangeKind_t::ALIVE,
reinterpret_cast<const uint8_t *>(&value),
static_cast<rtps::DataSize_t>(sizeof(value)));
logger.info("publish {} -> {}", value, change != nullptr ? "queued" : "dropped");
std::this_thread::sleep_for(std::chrono::milliseconds(period_ms));
}
Comment thread components/thread_pool/README.md Outdated
- Configurable worker count
- Bounded or unbounded queue
- Optional blocking submit mode for backpressure
- Graceful stop (drains queued jobs)
Comment thread .gitmodules
Comment on lines +40 to +42
[submodule "components/rtps_embedded/thirdparty/Micro-CDR"]
path = components/rtps_embedded/thirdparty/Micro-CDR
url = git@github.com:esp-cpp/Micro-CDR.git
Comment on lines +1 to +27
idf_component_register(
SRCS
"src/ThreadPool.cpp"
"src/communication/EsppTransport.cpp"
"src/discovery/ParticipantProxyData.cpp"
"src/discovery/SEDPAgent.cpp"
"src/discovery/SPDPAgent.cpp"
"src/discovery/TopicData.cpp"
"src/entities/Domain.cpp"
"src/entities/Participant.cpp"
"src/entities/Reader.cpp"
"src/entities/StatelessReader.cpp"
"src/entities/Writer.cpp"
"src/messages/MessageReceiver.cpp"
"src/messages/MessageTypes.cpp"
"src/utils/Diagnostics.cpp"
"thirdparty/Micro-CDR/src/c/common.c"
"thirdparty/Micro-CDR/src/c/types/array.c"
"thirdparty/Micro-CDR/src/c/types/basic.c"
"thirdparty/Micro-CDR/src/c/types/sequence.c"
"thirdparty/Micro-CDR/src/c/types/string.c"
INCLUDE_DIRS
"include"
"thirdparty/Micro-CDR/include"
REQUIRES
base_component cdr task thread_pool socket
)
Comment on lines +1 to +4
idf_component_register(
INCLUDE_DIRS "include"
SRC_DIRS "src"
REQUIRES base_component task)
@guo-max
guo-max marked this pull request as draft August 3, 2026 13:34
Copilot AI review requested due to automatic review settings August 3, 2026 13:35

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 72 out of 73 changed files in this pull request and generated no new comments.

Suppressed comments (7)

components/rtps_embedded/include/rtps/common/Locator.h:96

  • FullLengthLocator::serializeIntoUdcrBuffer is declared to return bool but does not return a value on the success path, which can lead to undefined return values (and may fail builds with -Werror).
  bool serializeIntoUdcrBuffer(ucdrBuffer &buffer) {
    if (ucdr_buffer_remaining(&buffer) < sizeof(FullLengthLocator)) {
      return false;

pc/tests/rtps_embedded_publisher.cpp:20

  • This file uses std::sscanf, but it never includes <cstdio>. With the current include set, std::sscanf is not declared and this test won't compile reliably.
    .gitmodules:42
  • The Micro-CDR submodule URL is using an SSH form (git@github.com:...), which will break anonymous CI checkouts and developers without configured SSH keys. Prefer an HTTPS URL here.
[submodule "components/rtps_embedded/thirdparty/Micro-CDR"]
	path = components/rtps_embedded/thirdparty/Micro-CDR
	url = git@github.com:esp-cpp/Micro-CDR.git

components/rtps_embedded/src/entities/Domain.cpp:118

  • User-traffic multicast forwarding filters participants using packet.destAddr, but the receive pipeline populates that field with the sender address (see ThreadPool::onDatagram / EsppTransport::onReceive). This means user multicast packets will typically not be delivered to any participant whose reader is bound to a multicast group.
    components/rtps_embedded/example/main/main.cpp:50
  • copy_len is computed as a bounded size, but copyInto is still called with sizeof(buffer). Since copyInto requires destSize >= change.getDataSize(), any message larger than the buffer will be silently dropped and copy_len becomes misleading. Either explicitly reject oversized payloads with a clear log or allocate a buffer sized to change.getDataSize().
  const rtps::DataSize_t copy_len =
      (change.getDataSize() < static_cast<rtps::DataSize_t>(sizeof(buffer) - 1))
          ? change.getDataSize()
          : static_cast<rtps::DataSize_t>(sizeof(buffer) - 1);

  if (copy_len == 0 ||
      !change.copyInto(reinterpret_cast<uint8_t *>(buffer), sizeof(buffer))) {
    return;

components/rtps_embedded/example/main/Kconfig.projbuild:8

  • The help text says the example verifies "UInt32 sample exchange", but the example code publishes/receives std_msgs::msg::String (see example/main/main.cpp). This is confusing when configuring the demo.
            Build one board as the initiator and a second board as the responder
            to verify RTPS discovery and end-to-end UInt32 sample exchange.

components/rtps_embedded/include/rtps/storages/CacheChange.h:67

  • CacheChange::reset() clears metadata but leaves the payload buffer intact, so large samples can stay resident in memory even after the change is "reset" (e.g., when history slots are reclaimed). Clearing the payload helps avoid unbounded memory retention on long runs.

Copilot AI review requested due to automatic review settings August 3, 2026 14:31

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 71 out of 72 changed files in this pull request and generated 2 comments.

Suppressed comments (9)

pc/tests/rtps_embedded_publisher.cpp:9

  • parse_ipv4 uses std::sscanf, but this file does not include <cstdio>, which can fail to compile on stricter toolchains.
    components/rtps_embedded/include/rtps/common/Locator.h:98
  • FullLengthLocator::serializeIntoUdcrBuffer does not return true on the success path, which will trigger “control reaches end of non-void function” (and can be treated as an error).
  bool serializeIntoUdcrBuffer(ucdrBuffer &buffer) {
    if (ucdr_buffer_remaining(&buffer) < sizeof(FullLengthLocator)) {
      return false;
    } else {
      ucdr_serialize_array_uint8_t(&buffer, reinterpret_cast<uint8_t *>(this),

.gitmodules:42

  • The Micro-CDR submodule URL uses an SSH-style remote (git@github.com:...). This will fail for users/CI environments without GitHub SSH keys; HTTPS is the more portable default for public submodules.
[submodule "components/rtps_embedded/thirdparty/Micro-CDR"]
	path = components/rtps_embedded/thirdparty/Micro-CDR
	url = git@github.com:esp-cpp/Micro-CDR.git

components/rtps_embedded/src/entities/Participant.cpp:170

  • deleteWriter() dereferences m_writers[i] without checking for nullptr, and also doesn't handle a null writer argument. This can crash when the array contains empty slots.
    components/rtps_embedded/src/entities/Domain.cpp:442
  • Same overflow issue as createWriter(): strlen(...) == MAX_*_LENGTH passes the current check but strcpy will overflow by writing the null terminator. Use >= and bounded copies.
    pc/tests/rtps_embedded_subscriber.cpp:70
  • The subscriber defaults to topic rtps_emb_pub but uses a String type (std_msgs::msg::String). The publisher test also defaults to rtps_emb_pub but publishes a UInt32, so running both with defaults will not interoperate and is confusing.
    components/rtps_embedded/include/rtps/discovery/ParticipantProxyData.h:169
  • Duration_t.fraction is encoded as seconds / 2^32 (see sysFunctions.h), but isAlive() treats it as microseconds by multiplying by 1e-6. This inflates lease durations and breaks liveliness expiration logic.
bool ParticipantProxyData::isAlive() const {
  uint32_t lease_in_ms =
      m_leaseDuration.seconds * 1000 + m_leaseDuration.fraction * 1e-6;

  uint32_t max_lease_in_ms =
      Config::SPDP_MAX_REMOTE_LEASE_DURATION.seconds * 1000 +
      Config::SPDP_MAX_REMOTE_LEASE_DURATION.fraction * 1e-6;

components/rtps_embedded/src/communication/EsppTransport.cpp:212

  • joinMultiCastGroup() returns success for the first multicast group even if all existing sockets fail to join (because m_multicastGroups.size() == 1). That can hide real join failures when channels already exist.
    components/rtps_embedded/src/entities/Participant.cpp:159
  • deleteReader() dereferences m_readers[i] without checking for nullptr, and also doesn't handle a null reader argument. This can crash when deleting readers if the array contains empty slots.

This issue also appears on line 161 of the same file.

Comment on lines +370 to +375
if (strlen(topicName) > Config::MAX_TOPICNAME_LENGTH ||
strlen(typeName) > Config::MAX_TYPENAME_LENGTH) {
return nullptr;
}
strcpy(attributes.topicName, topicName);
strcpy(attributes.typeName, typeName);
Comment on lines +26 to +33
#include "rtps/ThreadPool.h"

#include "rtps/entities/Domain.h"
#include "rtps/entities/Writer.h"
#include "rtps/utils/Diagnostics.h"
#include "rtps/utils/Log.h"
#include "rtps/utils/udpUtils.h"
#include "thread_pool.hpp"
Copilot AI review requested due to automatic review settings August 4, 2026 21:07
Comment thread components/rtps_embedded/example/host_pubsub.py Fixed
Comment thread components/rtps_embedded/example/host_pubsub.py Fixed

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 73 out of 74 changed files in this pull request and generated 2 comments.

Suppressed comments (6)

components/rtps_embedded/include/rtps/common/Locator.h:101

  • FullLengthLocator::serializeIntoUdcrBuffer() can reach the end of a non-void function without returning a value, which will fail compilation under stricter warnings/settings (and is undefined behavior if used).
    } else {
      ucdr_serialize_array_uint8_t(&buffer, reinterpret_cast<uint8_t *>(this),
                                   sizeof(FullLengthLocator));
    }
  }

pc/tests/rtps_embedded_publisher.cpp:9

  • This file uses std::sscanf and std::string but does not include / . This can fail to compile depending on transitive includes.
    .gitmodules:42
  • The Micro-CDR submodule URL uses an SSH form (git@github.com:...), which typically fails in CI/for contributors without GitHub SSH keys configured. Prefer an https:// URL for public submodules.
[submodule "components/rtps_embedded/thirdparty/Micro-CDR"]
	path = components/rtps_embedded/thirdparty/Micro-CDR
	url = git@github.com:esp-cpp/Micro-CDR.git

components/rtps_embedded/include/rtps/storages/CacheChange.h:67

  • CacheChange::reset() does not clear the PayloadBuffer, so caches that reuse CacheChange instances can retain large payload allocations indefinitely (and can keep stale data around longer than intended).
    pc/tests/rtps_embedded_subscriber.cpp:11
  • This file uses std::string and std::mutex but does not include / . Relying on transitive includes is brittle and can break compilation.
    components/rtps_embedded/README.md:8
  • This PR adds a new component, but the repo’s component conventions appear incomplete here: there is no components/rtps_embedded/idf_component.yml, and the docs/CI registries don’t reference rtps_embedded (no entries in doc/Doxyfile, .github/workflows/build.yml matrix, or .github/workflows/upload_components.yml). Without these, docs generation, CI example builds, and component-registry uploads will likely miss/fail this component.
# 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.


reference operator*() const { return m_pool.m_data[m_bit]; }

reference operator->() const { return m_pool.m_data[m_bit]; }
Comment on lines +245 to +249
if (valid_locators == Config::SPDP_MAX_NUM_LOCATORS) {
buffer.iterator += sizeof(FullLengthLocator);
PPD_LOG("Max number of valid locators exceeded, ignoring this locator as we have at least one valid locator");
return true;
}
Copilot AI review requested due to automatic review settings August 4, 2026 21:21

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 73 out of 74 changed files in this pull request and generated 2 comments.

Suppressed comments (10)

pc/tests/rtps_embedded_publisher.cpp:1

  • This file uses std::sscanf and std::string but does not include and . Relying on transitive includes is fragile and can break builds on different standard library implementations; add the missing standard headers explicitly.
    pc/tests/rtps_embedded_subscriber.cpp:1
  • This file uses std::mutex and std::string but does not include and . It may compile depending on transitive includes today, but can fail on other platforms/toolchains; include the required headers explicitly.
    components/rtps_embedded/src/messages/MessageReceiver.cpp:1
  • processMessage() ignores the return value of processSubmessage(). As written, processing failures (including unsupported submessages where processSubmessage returns false) will be silently ignored and processMessage will still return true, which can lead to hard-to-debug dropped/partial parsing. Make processMessage propagate failure (or have processSubmessage always return true for 'skipped but OK' cases).
    components/rtps_embedded/include/rtps/storages/CacheChange.h:1
  • CacheChange::reset() does not clear the payload buffer (data). Since HistoryCacheWithDeletion::incrementTail() calls reset() when dropping entries, payload vectors can retain (and potentially grow) capacity over time, increasing memory usage. Consider calling data.reset() (or otherwise clearing/shrinking as appropriate) as part of reset(), or documenting that payload memory retention is intentional.
    components/rtps_embedded/src/discovery/ParticipantProxyData.cpp:1
  • This branch advances ucdrBuffer by mutating buffer.iterator directly without updating the buffer bookkeeping (e.g., last_data_size/alignment). Other parts of the file use ucdr_advance_buffer() for safe skipping. Use ucdr_advance_buffer(&buffer, sizeof(FullLengthLocator)) here too to avoid deserialization state corruption.
    components/rtps_embedded/include/rtps/storages/ThreadSafeCircularBuffer.h:1
  • The method name peakFirst appears to be a typo for peekFirst (peek at the first element without removing it). If this is a new API surface, consider correcting the name now to avoid long-term public API drift.
    components/rtps_embedded/include/rtps/utils/udpUtils.h:1
  • The header guard is RTPS_UDP_UTILS_H but the trailing #endif comment says RTPS_UDP_H. Keeping these consistent reduces confusion during debugging and maintenance.
    components/rtps_embedded/src/communication/EsppTransport.cpp:1
  • New channels force the underlying UDP socket log level to DEBUG. This can produce very noisy logs and degrade runtime performance in deployments. Consider defaulting this to WARN/INFO (or making it configurable via rtps::Config) and only enabling DEBUG when explicitly requested.
    .gitmodules:42
  • The submodule URL uses the SSH form (git@github.com:...), which requires developers/CI environments to have SSH keys configured. For broader accessibility (and fewer CI failures), prefer an HTTPS URL (https://github.com/...).
[submodule "components/rtps_embedded/thirdparty/Micro-CDR"]
	path = components/rtps_embedded/thirdparty/Micro-CDR
	url = git@github.com:esp-cpp/Micro-CDR.git

components/rtps_embedded/include/rtps/discovery/TopicData.h:68

  • The default constructor hard-codes a non-local IP (192.168.0.42) into unicastLocator. If any code path accidentally uses a default-constructed TopicData before overriding locators, it can advertise invalid endpoints on the wire. Prefer initializing locators as invalid/zero (or using the local interface IP) to make unsafe defaults less likely.
  TopicData()
      : endpointGuid(GUID_UNKNOWN), typeName{'\0'}, topicName{'\0'},
        reliabilityKind(ReliabilityKind_t::BEST_EFFORT),
        durabilityKind(DurabilityKind_t::VOLATILE) {
    rtps::FullLengthLocator someLocator =
        rtps::FullLengthLocator::createUDPv4Locator(
            192, 168, 0, 42, rtps::getUserUnicastPort(0));
    unicastLocator = someLocator;
    multicastLocator = FullLengthLocator();
  };

Comment on lines +94 to +101
bool serializeIntoUdcrBuffer(ucdrBuffer &buffer) {
if (ucdr_buffer_remaining(&buffer) < sizeof(FullLengthLocator)) {
return false;
} else {
ucdr_serialize_array_uint8_t(&buffer, reinterpret_cast<uint8_t *>(this),
sizeof(FullLengthLocator));
}
}
Comment on lines +163 to +171
bool ParticipantProxyData::isAlive() const {
uint32_t lease_in_ms =
m_leaseDuration.seconds * 1000 + m_leaseDuration.fraction * 1e-6;

uint32_t max_lease_in_ms =
Config::SPDP_MAX_REMOTE_LEASE_DURATION.seconds * 1000 +
Config::SPDP_MAX_REMOTE_LEASE_DURATION.fraction * 1e-6;

auto heatbeat_age_in_ms = getAliveSignalAgeInMilliseconds();
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants