Two domain participants on the same domain in the same application do not communicate unless run in the debugger.

[wcneill]

I have written DDSNode class, which is a wrapper class for a DomainParticipant. It bundles all of the DDS communication logic needed for a containing class or application.

When I create a single one of these nodes, it can communicate with itself via its publisher and subscriber. It writes some data out, and reads that data back in.

However, when I create a second FastDDSNode containing a new domain participant within the same application, the two cannot communicate unless I run the application in the debugger.

I'm not sure the best order in which to present the relevant code, but here's my best shot:

1. First, I'll show you the code that sets up the publishers/subscribers/readers/writers for this FastDDSNode class

2. Then, I'll show you the type erased wrapper class for writing generic IDL generated data types, and how FastDDSNode.read() and FastDDSNode.write() work with this type erased class.

3. Finally, I will show you some simple unit tests I have written that verify a single domain participant can communicate with itself, but not with other domain participants subscribed to the same topics.

1. FastDDSNode wrapper class

class FastDDSNode : public IDDSInterface
{

private:
    eprosima::fastdds::dds::DomainParticipant *participant_;
    eprosima::fastdds::dds::Publisher *publisher_;
    eprosima::fastdds::dds::Subscriber *subscriber_;

    std::map<std::string, eprosima::fastdds::dds::DataReader*> reader_map_;
    std::map<std::string, eprosima::fastdds::dds::DataWriter*> writer_map_;
    std::map<std::string, std::pair<eprosima::fastdds::dds::Topic*, eprosima::fastdds::dds::TypeSupport>> topic_map_;

public:
   // Constructor.... 
    FastDDSNode(
            const std::string &topic_name
            , bool publisher
            , bool subscriber
            , std::string node_name
            , eprosima::fastdds::dds::TypeSupport type_support
            , eprosima::fastdds::dds::DomainParticipantQos domain_qos = eprosima::fastdds::dds::PARTICIPANT_QOS_DEFAULT
            , eprosima::fastdds::dds::SubscriberQos sub_qos = eprosima::fastdds::dds::SUBSCRIBER_QOS_DEFAULT
            , eprosima::fastdds::dds::PublisherQos pub_qos = eprosima::fastdds::dds::PUBLISHER_QOS_DEFAULT
            , eprosima::fastdds::dds::TopicQos topic_qos = eprosima::fastdds::dds::TOPIC_QOS_DEFAULT
            , eprosima::fastdds::dds::DataWriterQos writer_qos = eprosima::fastdds::dds::DATAWRITER_QOS_DEFAULT
            , eprosima::fastdds::dds::DataReaderQos reader_qos = eprosima::fastdds::dds::DATAREADER_QOS_DEFAULT
        ) : participant_(nullptr), publisher_(nullptr), subscriber_(nullptr)
        {
            using namespace eprosima::fastdds::dds;
            LOG_INFO("FastDDSNode: Initializing for topic: " + topic_name);
    
            // Create the DomainParticipant
            DomainParticipantQos participantQos;
            participantQos.name(node_name);
            participant_ = DomainParticipantFactory::get_instance()->create_participant(0, participantQos);
            if (!participant_)
                throw std::runtime_error("Could not create DomainParticipant");
    
    
           // Verify TypeSupport is valid
            if (type_support.empty())
            {
                LOG_ERROR("FastDDSNode: TypeSupport is empty for topic: " + topic_name);
                throw std::runtime_error("TypeSupport is empty");
            }
    
            // Register TypeSupport
            type_support.register_type(participant_);
    
            // Create and verify topic
            Topic* topic_ptr = participant_->create_topic(topic_name, type_support.get_type_name(), topic_qos);
            if (!topic_ptr) 
            {
                LOG_ERROR("FastDDSNode: Could not create Topic: " + topic_name);
                throw std::runtime_error("Could not create Topic");
            }
            
            std::pair<Topic*, TypeSupport> topic_pair(topic_ptr, type_support);
            topic_map_[topic_name] = topic_pair;
    
            if (publisher)
            {
                publisher_ = participant_->create_publisher(pub_qos, nullptr);
                writer_map_[topic_name] = publisher_->create_datawriter(topic_ptr, writer_qos, nullptr);
            }
            if (subscriber)
            {
                subscriber_ = participant_->create_subscriber(sub_qos, nullptr);
                reader_map_[topic_name] = subscriber_->create_datareader(topic_ptr, reader_qos, nullptr);
            }
    
            LOG_INFO("FastDDSNode: Initialized for topic: " + topic_name);
        }
    )

// ... reader/writer methods shown below in next section. 

};

2. TypeErasure for reading/writing generic data with a single read/write method

// Type erasure games, so a DDS node can handle any data type
class DDSDataWrapperBase 
{
    public:
        virtual ~DDSDataWrapperBase() = default;
        virtual void* ptr() = 0;
        virtual std::type_index type() const = 0;
};

template <typename T>
class DDSDataWrapper : public DDSDataWrapperBase 
{

public:
    DDSDataWrapper(T& data) : data_(data) {}

    /**
     * @brief Returns a pointer to the internal data. 
     */
    void* ptr() override {
        return static_cast<void*>(&data_);
    }

    std::type_index type() const override {
        return typeid(T);
    }

    const T& data() const {
        return data_;
    }

private:
    T data_;
};

And here are the read/write methods that utilize the type erasure classes:

 /**
     * @brief Publishes a sample to the DDS topic.
     * @param data The topic instance to publish.
     * @return True on success, false otherwise.
     */
    bool write(const std::string& topic_name, DDSDataWrapperBase &msg) override
    {

        using namespace eprosima::fastdds::dds;

        auto it = writer_map_.find(topic_name);
        if (it == writer_map_.end()) {
            LOG_ERROR("No data writer found for " + topic_name);
            return false; // Writer not found for the topic
        }

        DataWriter* writer = it->second;

        if(!writer) 
        {
            LOG_ERROR("Null pointer found in place of data writer for topic " + topic_name);
            return false;
        }

        ReturnCode_t code = writer->write(msg.ptr());
        if (code != RETCODE_OK)
        {
            LOG_ERROR("DataWriter failed with code: " + code);
            return false;
        }
        
        LOG_INFO("Message successfully written.");
        return true;
    }

    /**
     * @brief Reads a sample from the DDS topic (if available).
     * @param out_data The output data instance to fill with the received sample.
     * @return True if a valid sample was read, false otherwise.
     */
    bool read(const std::string& topic_name, DDSDataWrapperBase& out_data) override
    {
        using namespace eprosima::fastdds::dds;

        auto it = reader_map_.find(topic_name);
        if (it == reader_map_.end()) {
            LOG_ERROR("FastDDSNode: Reader not found for topic: " + topic_name);
            return false; // Reader not found for the topic
        }

        DataReader *reader = it->second;

        if(!reader) {
            LOG_ERROR("Null pointer found in place of DataReader for topic" + topic_name);
            return false;
        }

        SampleInfo info;
        ReturnCode_t return_code = reader->take_next_sample(out_data.ptr(), &info);
        bool is_read = return_code == RETCODE_OK;
        bool is_valid = info.valid_data;
        if (is_read && is_valid)
        {
            return true;
        }
        return false;
    }

3. Unit tests, written with GTest

The following unit test will pass in the debugger, but fail if I just run the compiled executable.

The important note is that DataReader->take_next_sample() returns RETCODE_OK and info.valid_data == true... but on inspection, the read data only contains default values rather than the ones written by the DataWriter.

class NodeFixture : public testing::Test
{
    protected:
        std::string topic_name;
        TypeSupport type_support;
        FastDDSNode node_a;
        FastDDSNode node_b;
        SensorECS::SensorData sensor_data_out;
        SensorECS::DeviceStatus device_data_out;


        NodeFixture() 
            : topic_name("TestSensorTopic")
            , type_support(new SensorECS::SensorDataPubSubType)
            , node_a(topic_name, true, true, "node_a", type_support)
            , node_b(topic_name, true, true, "node_b", type_support)
        {

            // SensorData structure
            sensor_data_out.base().timestamp_us(123456789);
            sensor_data_out.base().uid("unit_test_uid");
            sensor_data_out.base().source_id("unit_test_source");

            std::string json_str = "{\"test\":\"value\"}";
            sensor_data_out.base().json_data(json_str);
            sensor_data_out.temperature(25.5f);
            sensor_data_out.humidity(60.0f);

            // DeviceData structure
            device_data_out.base().timestamp_us(123456789);
            device_data_out.base().uid("another_uid");
            device_data_out.base().source_id("another_source");
            device_data_out.is_online(true);
            device_data_out.status_message("I hope your tests pass.");

        }

};

// Test creating a DDS node as publisher and writing a sample
TEST_F(NodeFixture, BaseNodeReadAndWrite)
{
    try
    {
        DDSDataWrapper<SensorECS::SensorData> out_data_wrapper(sensor_data_out);
        bool success = node_a.write("TestSensorTopic", out_data_wrapper);
        EXPECT_TRUE(success);

        std::this_thread::sleep_for(std::chrono::seconds(10));
        
        SensorECS::SensorData sensor_data_in;
        DDSDataWrapper<SensorECS::SensorData> in_data_wrapper(sensor_data_in);
        success = node_b.read("TestSensorTopic", in_data_wrapper);  
        EXPECT_TRUE(success);
        EXPECT_EQ(in_data_wrapper.data().humidity(), sensor_data_out.humidity());
    }
    catch (const std::exception &e)
    {
        FAIL() << "Exception thrown: " << e.what();
    }

Additionally, if you alter the test so that node_a does both the writing AND reading, it will pass.

Simply this line

success = node_b.read("TestSensorTopic", in_data_wrapper);

to this:

success = node_a.read("TestSensorTopic", in_data_wrapper);

[wcneill]

A couple other things I've messed with:

1. I have verified that the tests only pass in the debugger if I insert a break point, and then continue.
2. I have also added a DataReaderListener to node_b on the topic... The behavior is the same. It work in the debugger but communication fails when simply running the compiled executable.
3. I read about the idea of SharedMemory transport in the user's guide... it says it is enabled by default and automatically used if two participants are on the same machine... but I manually enabled it just in case. Same behavior.

[wcneill]

I also tested just using FastDDS with none of my wrapper code. The issue remains.

But, I figured it out... I had a wait between write() and read() in my tests. Turns out what I really needed was a wait between Reader/Writer creation and the first Write/Read:

// Test adding new topics, readers and writers 
TEST_F(NodeFixture, AddNewTopicReadAndWrite)
{
    std::string new_topic_name = "DeviceStatusTopic";
    try
    {
        // Test add new topic
        bool success = true;
        TypeSupport device_status_support(new SensorECS::DeviceStatusPubSubType());
        success = node_a.add_topic(new_topic_name, device_status_support);
        EXPECT_TRUE(success) << "Failed to add new topic to node" << std::endl;

        // Test add writer for new topic
        success = node_a.add_writer(new_topic_name); 
        EXPECT_TRUE(success) << "Failed to add new writer" << std::endl;

        // Test add reader for new topic
        node_b.add_topic(new_topic_name, device_status_support);
        success = node_b.add_reader(new_topic_name);
        EXPECT_TRUE(success) << "Failed to add new reader." << std::endl;

        // wait a second for discovery to occur
        std::this_thread::sleep_for(std::chrono::seconds(1));  //<<<<<<<<<<<<<<<<<<<<<<<<< HAD TO WAIT HERE

        // Test write to new topic
        DDSDataWrapper<SensorECS::DeviceStatus> out_data_wrapper(device_data_out);
        success = node_a.write(new_topic_name, out_data_wrapper);
        EXPECT_TRUE(success) << "Failed to write device status" << std::endl;

        // Test read from new topic
        SensorECS::DeviceStatus device_data_in;
        DDSDataWrapper<SensorECS::DeviceStatus> in_data_wrapper(device_data_in);
        success = node_b.read(new_topic_name, in_data_wrapper);
        EXPECT_TRUE(success) << "New reader failed to read" << std::endl;
        EXPECT_EQ(in_data_wrapper.data().status_message(), out_data_wrapper.data().status_message());
        EXPECT_EQ(in_data_wrapper.data().is_online(), out_data_wrapper.data().is_online());
    }
    catch (const std::exception &e)
    {
        FAIL() << "Exception thrown " << e.what();
    }
}

[MiguelCompany]

The default durability settings imply that a reader will not receive samples written by the writer before it discovered the reader.

Setting the reader as TRANSIENT_LOCAL, and the reliability as RELIABLE would allow removing the wait.

Another option is to wait for the writer to discover the reader before writing. You could use on_publication_matched for that

[wcneill]

Hi Miguel, setting the durability and reliability settings did the trick!

Thank you, kind sir.