fast-rtps subscriber issue

listener.cpp

bool Listener::init(Participant *p)
{
    // set up subscription for Latency msgs
    SubscriberAttributes sa;
    sa.topic.topicKind = NO_KEY;
    sa.topic.topicDataType = "TestTopicType";
    sa.topic.topicName = "HelloTopic";
    // sa.qos.m_reliability.kind = BEST_EFFORT_RELIABILITY_QOS;
    sa.qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;

    bool dropnew = true;
    int depth = 2;

    if (dropnew) {
        sa.topic.historyQos.kind = eprosima::fastrtps::KEEP_ALL_HISTORY_QOS;
        sa.topic.resourceLimitsQos.max_samples = depth;
        sa.topic.historyQos.depth = depth;
    } else {
        sa.topic.historyQos.kind = eprosima::fastrtps::KEEP_LAST_HISTORY_QOS;
        sa.topic.resourceLimitsQos.max_samples = depth;
        sa.topic.historyQos.depth = depth;
    }

    subscriber = Domain::createSubscriber(p, sa, this);
    if (subscriber == nullptr) {
        return false;
    }

    return true;
}

void Listener::readOne()
{
    TestTopicData tt;
    while (subscriber->takeNextData((void *)&tt, &sampleInfo)) {
        if (sampleInfo.sampleKind == ALIVE) {
            n_samples++;
            // Print your structure data here.
            cout << tt.seqnum << " RECEIVED" << endl;
        }
    }
}

void Listener::run()
{
    for (;;) {
        sleep(1);
        listener.readOne();
    }
}

talker.cpp

bool Talker::init(Participant *p)
{
    PublisherAttributes pa;
    pa.topic.topicDataType = "TestTopicType";
    pa.topic.topicKind = NO_KEY;
    pa.topic.topicName = "HelloTopic";
    // pa.qos.m_reliability.kind = BEST_EFFORT_RELIABILITY_QOS;
    pa.qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;

    pa.times.heartbeatPeriod.fraction = 4294967 * 200; // ~200 millis

    // pa.topic.historyQos.kind = eprosima::fastrtps::KEEP_ALL_HISTORY_QOS;
    pa.topic.historyQos.kind = eprosima::fastrtps::KEEP_LAST_HISTORY_QOS;
    pa.topic.historyQos.depth = 20;
    pa.topic.resourceLimitsQos.max_samples = 20;

    mp_datapub = Domain::createPublisher(p, pa, &m_datapublistener);
    if (mp_datapub == nullptr) {
        return false;
    }

    return true;
}

void Talker::run()
{
    unsigned count = 0;
    
    for (;;) {
        publish();

        if (count++ < 2) {
            eClock::my_sleep(100);
        } else {
            eClock::my_sleep(2000);
        }
    }
}

bool Talker::publish()
{
    if (m_datapublistener.n_matched <= 0) {
        return false;
    }

    ttd.seqnum++;
    cout << "talker:" << ttd.seqnum << endl;
    return mp_datapub->write((void *)&ttd);
}