How to make sure that a subscriber actually received the publisher's message?

[harsszegi]

Hi,

I'm using RELIABLE_RELIABILITY_QOS for both publisher and subscriber over WAN + TCPv4. I have an issue. My publisher client code should be a "fire and forget" client, meaning it sends out the message, then it should close the channel if the subscriber has received the message.

When DataWriter's write() returns I suppose the message has been pushed onto the TCP socket, however as I close the socket via closing the writer, publisher, topic, etc., the subscriber is not receiving it.

Obviously If I "wait enough" after doing the write() and before closing the socket everything works just fine.

Is there a QOS or any other solution that I could use for the publisher to make sure that subscriber actually read the message from the socket?
Thanks,

[JLBuenoLopez]

Hi @harsszegi

You might be interested on the DataWriter::wait_for_acknowledgments API defined in Clause 2.2.2.4.2.15 of the DDS specification.

[harsszegi]

Funny thing though, that I'm actually doing this :(

  std::vector<DataWriter *> writers;

  size_t n = 10;
  PubListener listener;

  for (auto i = 0; i < n; i++)
  {
    DataWriterQos wqos;
    wqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
    wqos.history().kind = KEEP_LAST_HISTORY_QOS;
    wqos.history().depth = 1;
    wqos.resource_limits().max_samples = 50;
    wqos.resource_limits().max_samples_per_instance = 20;
    wqos.reliable_writer_qos().times.heartbeatPeriod.seconds = 2;
    wqos.reliable_writer_qos().times.heartbeatPeriod.nanosec = 200 * 1000 * 1000;
    auto writer = publisher->create_datawriter(topic, wqos, &listener);
    if (writer == nullptr)
    {
      std::cerr << "Unable to create writer" << std::endl;
      return 1;
    }

    writers.push_back(writer);
  }

  auto action = [&](size_t i) {
    if (!writers[i]->write(&message))
    {
      std::cerr << "Unable to send message" << std::endl;
    }

    if (ReturnCode_t::RETCODE_OK != writers[i]->wait_for_acknowledgments(eprosima::fastrtps::c_TimeInfinite))
    {
      std::cerr << "Message is not acknowledged" << std::endl;
    }
  };

Then I simply call all lambdas. After executing the lambdas, I'm deleting every resource, e.g:

  for (auto w : writers)
  {
    publisher->delete_datawriter(w);
  }
  participant->delete_publisher(publisher);
  participant->delete_topic(topic);
  DomainParticipantFactory::get_instance()->delete_participant(participant);

So even with the wait_for_acknowledgement if I execute the resource freeing code, the messages are not received.

[JLBuenoLopez]

Have you ensured that your DataReader have discovered and matched your DataWriter before sending the sample? If you write the sample directly and there is no DataReader matched, when the match occurs, the DataReader is VOLATILE by default, so the past samples are not resent. Also, if you have destroyed the DataWriter, the sample would also be removed. If there is no DataReader matched, DataWriter::wait_for_acknowledgments would return directly as there is no one to send the sample.

[harsszegi]

This is very strange, I have changed the code in a way that the matched writers are being stored in a global vector, and then the lambda will use this global vector and before calling the lambda I make sure that the global vector is filled properly the all the matched writers.

What I can see that all writers are matched, I am sending out 160k 10x, but the subscriber only received 3 of the messages.

Are there any thread-safety concerns if I define:

• one participant
• one topic
• one publisher
• one message
• create multiple writer on the same topic / publisher
• send the same message on the different writers

Is this a valid use case or should I crreate different participants / publishers as well?
Thanks,

[harsszegi]

Oh, using one writer works fine!