I am learning about DDS using RTI (still very new to this topic) . I am creating a Publisher that writes to a Subscriber, and the Subscriber outputs the message. One thing I would like to simulate is dropped packages. As an example, let's say the Publisher writes to the Subscriber 4 times a second but the Subscriber can only read one a second (the most recent message).
As of now, I am able to create a Publisher & Subscriber w/o any packages being dropped.
I read through some documentation and found HistoryQosPolicyKind.KEEP_LAST_HISTORY_QOS.
Correct me if I am wrong, but I was under the impression that this would essentially keep the most recent message received from the Publisher. Instead, the Subscriber is receiving all the messages but delayed by 1 second.
I don't want to cache the messages but drop the messages. How can I simulate the "dropped" package?
BTW: I don't want to change anything in the .xml file. I want to do it programmatically.
Here are some snippets of my code.
//Publisher.java
//writer = (MsgDataWriter)publisher.create_datawriter(topic, Publisher.DATAWRITER_QOS_DEFAULT,null /* listener */, StatusKind.STATUS_MASK_NONE);
writer = (MsgDataWriter)publisher.create_datawriter(topic, write, null,
StatusKind.STATUS_MASK_ALL);
if (writer == null) {
System.err.println("create_datawriter error\n");
return;
}
// --- Write --- //
String[] messages= {"1", "2", "test", "3"};
/* Create data sample for writing */
Msg instance = new Msg();
InstanceHandle_t instance_handle = InstanceHandle_t.HANDLE_NIL;
/* For a data type that has a key, if the same instance is going to be
written multiple times, initialize the key here
and register the keyed instance prior to writing */
//instance_handle = writer.register_instance(instance);
final long sendPeriodMillis = (long) (.25 * 1000); // 4 per second
for (int count = 0;
(sampleCount == 0) || (count < sampleCount);
++count) {
if (count == 11)
{
return;
}
System.out.println("Writing Msg, count " + count);
/* Modify the instance to be written here */
instance.message =words[count];
instance.sender = "some user";
/* Write data */
writer.write(instance, instance_handle);
try {
Thread.sleep(sendPeriodMillis);
} catch (InterruptedException ix) {
System.err.println("INTERRUPTED");
break;
}
}
//writer.unregister_instance(instance, instance_handle);
} finally {
// --- Shutdown --- //
if(participant != null) {
participant.delete_contained_entities();
DomainParticipantFactory.TheParticipantFactory.
delete_participant(participant);
}
//Subscriber
// Customize time & Qos for receiving info
DataReaderQos readerQ = new DataReaderQos();
subscriber.get_default_datareader_qos(readerQ);
Duration_t minTime = new Duration_t(1,0);
readerQ.time_based_filter.minimum_separation.sec = minTime.sec;
readerQ.time_based_filter.minimum_separation.nanosec = minTime.nanosec;
readerQ.history.kind = HistoryQosPolicyKind.KEEP_LAST_HISTORY_QOS;
readerQ.reliability.kind = ReliabilityQosPolicyKind.BEST_EFFORT_RELIABILITY_QOS;
reader = (MsgDataReader)subscriber.create_datareader(topic, readerQ, listener, StatusKind.STATUS_MASK_ALL);
if (reader == null) {
System.err.println("create_datareader error\n");
return;
}
// --- Wait for data --- //
final long receivePeriodSec = 1;
for (int count = 0;
(sampleCount == 0) || (count < sampleCount);
++count) {
//System.out.println("Msg subscriber sleeping for "+ receivePeriodSec + " sec...");
try {
Thread.sleep(receivePeriodSec * 1000); // in millisec
} catch (InterruptedException ix) {
System.err.println("INTERRUPTED");
break;
}
}
} finally {
// --- Shutdown --- //
On the subscriber side, it is useful to distinguish three different types of interaction between your application and the DDS Domain: polling, Listeners and WaitSets
Polling means that the application decides when it reads available data. This is often a time-driven mechanism.
Listeners are basically callback functions that get invoked as soon as data becomes available, by an infrastructure thread, to read that data.
WaitSets implement a mechanism similar to the socket select mechanism: an application thread waits (blocks) for data to become available and after unblocking reads the new data.
Your application uses a Listener mechanism. You did not post the implementation of the callback function, but from the overall picture, it is likely that the listener implementation immediately tries to read the data at the moment that the callback is invoked. There is no time for the data to be "pushed out" or "dropped" as you called it. This reading happens in a different thread than your main thread, which is sleeping most of the time. You can find a Knowledge Base article about it here.
The only thing that is not clear is the impact of the time_based_filter QoS setting. You did not mention that in your question, but it does show up in the code. I would expect this to filter out some of your samples. That is a different mechanism than the pushing out of the history though. The behavior for the time based filter may be implemented differently for different DDS implementations. Which product do you use?