DDS之DCPS Subscription模块

当Subscriber 接收到数据的时候，会通过它的Listener或者使能一些Condition通知到Application有一些新的数据是可获取的，然后Application就可以通过其相关的DataReader的操作函数去获取该数据。
关于Subscription模块的类图如下：

有关类图中的关系可以查看：UML类图中的几种关系
由以上类图可以看出:

TopicDescription是Topic的父类
DataReaderListener是SubscriberListener的父类，继承了其定义的Callback函数
ReadCondition是QueryCondition的父类
Subscriber可以包含DataReader, QosPolicy, StatusCondition, SubscriberListener
DataReader可以包含DataReaderListener，QosPolicy，ReadCondition, TopicDescription, Data, DataSample
DataReader可以关联或者包含多个DataSample，每一个DataSample Class有一个Data 和Sample Info
以上小节分别对其内容进行分析

Subscriber

从Subscriber这个类的定义来看，除了图1类图中的方法之外，它还会继承自Entity的方法，比如get_listener，set_listener，get_qos，set_qos

怎样创建Subscriber

从图1可以看出Subscriber需要调用DomainParticipant 的成员函数创建。

在domain中创建DomainParticipant

// Create a DomainParticipant in the desired domain
DomainParticipant* participant =DomainParticipantFactory::get_instance()->create_participant(0, PARTICIPANT_QOS_DEFAULT);
if (nullptr == participant)
{// Errorreturn;
}

创建CustomSubscriberQos

// A custom SubscriberQos can be provided to the creation method
SubscriberQos custom_qos;// Modify QoS attributes
custom_qos.entity_factory().autoenable_created_entities = false;

SubscriberQos用于控制Subscriber的行为，与Subscriber相关的Qos Policy 主要包含如下：

QosPolicy class	作用
PartitionQosPolicy
PresentationQosPolicy
GroupDataQosPolicy
EntityFactoryQosPolicy

创建CustomSubscriberListener

CustomSubscriberListener custom_listener;// Modify Listener attributes
// (...)

SubscriberListener 继承自DataReaderListener , 其主要作用是在Subscriber的状态变化或者DataReader的一些特定的Event被触发之后调用其定义的一些callback, 有点儿类似于Hook函数，默认情况下，这部分内容为空，用户使用需要自己定义其内容。

创建不同的Subscriber

// Create a Subscriber with default SubscriberQos and no Listener
// The value SUBSCRIBER_QOS_DEFAULT is used to denote the default QoS.
Subscriber* subscriber_with_default_qos =participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
if (nullptr == subscriber_with_default_qos)
{// Errorreturn;
}

Subscriber* subscriber_with_custom_qos =participant->create_subscriber(custom_qos);
if (nullptr == subscriber_with_custom_qos)
{// Errorreturn;
}

Subscriber* subscriber_with_default_qos_and_custom_listener =participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT, &custom_listener);
if (nullptr == subscriber_with_default_qos_and_custom_listener)
{// Errorreturn;
}

Subscriber其他属性的更改/获取

// Get the current QoS or create a new one from scratch
SubscriberQos qos = subscriber->get_qos();// Assign the new Qos to the object
subscriber->set_qos(qos1);// Get the current QoS or create a new one from scratch
SubscriberQos qos_type1 = participant->get_default_subscriber_qos();SubscriberQos qos_type2;
// Set as the new default SubscriberQos
if (participant->set_default_subscriber_qos(qos_type2) != ReturnCode_t::RETCODE_OK)
{// Errorreturn;
}

DataReader

怎样创建/删除DataReader

创建DataReader需要Subscriber的成员函数create_datareader，其函数的定义与返回值如下：

DataReader *create_datareader(TopicDescription *topic, const DataReaderQos &reader_qos, DataReaderListener *listener = nullptr, const StatusMask &mask = StatusMask::all())

Parameters： topic – Topic the DataReader will be listening（Mandatory）.
reader_qos – QoS of the DataReader（Mandatory）.
listener – Pointer to the listener (default: nullptr)（Optional ）
mask – StatusMask that holds statuses the listener responds to (default: all)（Optional ）.
Returns： Pointer to the created DataReader. nullptr if failed.

reader_qos可以用默认的DataReaderQos也就是DATAREADER_QOS_DEFAULT，也可以用自行创建一个Custom_Qos

A StatusMask that activates or deactivates triggering of individual callbacks on the DataReaderListener. By default all events are enabled.

创建相应的Subscriber
创建特定的DataReaderQos

// A custom DataReaderQos can be provided to the creation method
DataReaderQos custom_qos;// Modify QoS attributes
// (...)

创建特定的CustomDataReaderListener

// CustomDataReaderListener inherits from DataReaderListener.
CustomDataReaderListener custom_listener;

调用Subscriber的成员函数create_datareader创建，delete_datareader删除

DataReaderQos custom_qos;
CustomDataReaderListener custom_listener;
DataReader* data_reader_with_default_qos_and_custom_listener =subscriber->create_datareader(topic, DATAREADER_QOS_DEFAULT, &custom_listener);
if (nullptr == data_reader_with_default_qos_and_custom_listener)
{// Errorreturn;
}
// Delete the DataReader
if (subscriber->delete_datareader(data_reader_with_default_qos_and_custom_listener ) != ReturnCode_t::RETCODE_OK)
{// Errorreturn;
}

Access to the data

DataReader, SampleInfo, Instance, Sample之间的关系

首先需要搞清楚DataReader, SampleInfo, Instance, Sample之间的关系如下：

SampleInfo包含的属性如下表：

SampleInfo	Sample/Instance	Functionality
sample_state	sample	表示每个DataReader对每一个收到的Sample是否读取的标志
view_state	sample	表示收到的这个sample是否是其对应的Instance的第一个
instance_state	Instance	表示一个Instance的状态： ALIVE：目前还存在 NOT_ALIVE_DISPOSED：DataWriter 舍弃了此Instance NOT_ALIVE_NO_WRITERS：DataReader舍弃了此Instance因为远程DataWriter未发布此Instance
valid_data	sample	TRUE：Data sample包含数据和SampleInfo FALSE：Data sample只有SampleInfo
disposed_generation_count	Instance	表示Instance从NOT ALIVE *变成ALIVE的次数
no_writers_generation_count	instance	表示Instance从NOT_ALIVE_NO_WRITERS变成ALIVE的次数
sample_rank	Instance	表示针对此Instance目前还有多少个未读取的 Sample
generation_rank	Instance	表示从接收到一个Instance的sample到此Instance的最新sample（MRSIC）进入Collection，Instance从NOT ALIVE到ALIVE的次数
absolute_generation_rank	Instance	表示从接收到一个Instance的sample到接收此Instance的最新的sample（MRSIC）期间，Instance从NOT ALIVE到ALIVE的次数
source_timestamp	sample	DataWriter发布此sample的时间戳

DataReader怎么取值

目前看来DataReader获取数据主要有如下三种方式：

利用Read/Take 等DataReader方法
利用DataReaderListener callback
利用Wait-Set与Condition

利用Read/Take, SampleInfo获取数据

DataReader::read()函数原型如下：

ReturnCode_t read(LoanableCollection &data_values, SampleInfoSeq &sample_infos, int32_t max_samples = LENGTH_UNLIMITED, SampleStateMask sample_states = ANY_SAMPLE_STATE, ViewStateMask view_states = ANY_VIEW_STATE, InstanceStateMask instance_states = ANY_INSTANCE_STATE)

DataReader::take()函数原型如下：

ReturnCode_t take(LoanableCollection &data_values, SampleInfoSeq &sample_infos, int32_t max_samples = LENGTH_UNLIMITED, SampleStateMask sample_states = ANY_SAMPLE_STATE, ViewStateMask view_states = ANY_VIEW_STATE, InstanceStateMask instance_states = ANY_INSTANCE_STATE)

以上两个函数返回LoanableCollection 和SampleInfoSeq 数据类型，示例如下：

    // Sequences are automatically initialized to be empty (maximum == 0)FooSeq data_seq;SampleInfoSeq info_seq;// with empty sequences, a take() or read() will return loaned// sequence elementsReturnCode_t ret_code = data_reader->take(data_seq, info_seq,LENGTH_UNLIMITED, ANY_SAMPLE_STATE,ANY_VIEW_STATE, ANY_INSTANCE_STATE);// process the returned data// must return the loaned sequences when done processingdata_reader->return_loan(data_seq, info_seq);// process the returned dataif (ret_code == ReturnCode_t::RETCODE_OK){// Both info_seq.length() and data_seq.length() will have the number of samples returnedfor (FooSeq::size_type n = 0; n < info_seq.length(); ++n){// Only samples for which valid_data is true should be accessedif (info_seq[n].valid_data){// Do something with data_seq[n]}}// must return the loaned sequences when done processingdata_reader->return_loan(data_seq, info_seq);}

利用DataReaderListener获取数据

当DataReader收到其配对的DataWriter发送的数据之后，可以通过DataReaderListener的两个callback来通知应用。

inline virtual void on_data_available(DataReader *reader)
Virtual function to be implemented by the user containing the actions to be performed when a new Data Message is received.

class CustomizedDataReaderListener : public DataReaderListener
{public:CustomizedDataReaderListener(): DataReaderListener(){}virtual ~CustomizedDataReaderListener(){}virtual void on_data_available(DataReader* reader){// Create a data and SampleInfo instanceFoo data;SampleInfo info;// Keep taking data until there is nothing to takewhile (reader->take_next_sample(&data, &info) == ReturnCode_t::RETCODE_OK){if (info.valid_data){// Do something with the datastd::cout << "Received new data value for topic "<< reader->get_topicdescription()->get_name()<< std::endl;}else{std::cout << "Remote writer for topic "<< reader->get_topicdescription()->get_name()<< " is dead" << std::endl;}}}};

inline virtual void on_data_on_readers(Subscriber *sub)
Virtual function to be implemented by the user containing the actions to be performed when a new Data Message is available on any reader.

利用Wait-Set获取数据

通过应用起一个线程，利用Wait-Set等到相应的状态满足或者设定的Timeout之后就获取数据。
示例如下：

// Create a DataReader
DataReader* data_reader =subscriber->create_datareader(topic, DATAREADER_QOS_DEFAULT);
if (nullptr == data_reader)
{// Errorreturn;
}// Prepare a wait-set to wait for data on the DataReader
WaitSet wait_set;  // Create a WaitSet instance
StatusCondition& condition = data_reader->get_statuscondition();//定义DataReader的StatusCondition
condition.set_enabled_statuses(StatusMask::data_available());//设置StatusCondition的trigger_value为Communication Status的DATA_AVAILABLE
wait_set.attach_condition(condition);//将定义的Condition添加到wait_set// Create a data and SampleInfo instance
Foo data;
SampleInfo info;//Define a timeout of 5 seconds
eprosima::fastrtps::Duration_t timeout (5, 0);// Loop reading data as it arrives
// This will make the current thread to be dedicated exclusively to
// waiting and reading data until the remote DataWriter dies
while (true)
{ConditionSeq active_conditions;if (ReturnCode_t::RETCODE_OK == wait_set.wait(active_conditions, timeout)){while (ReturnCode_t::RETCODE_OK == data_reader->take_next_sample(&data, &info)){if (info.valid_data) //如果sample中的Data部分有内容{// Do something with the datastd::cout << "Received new data value for topic "<< topic->get_name()<< std::endl;}else{// If the remote writer is not alive, we exit the reading loopstd::cout << "Remote writer for topic "<< topic->get_name()<< " is dead" << std::endl;break;}}}else{std::cout << "No data this time" << std::endl;}
}