OpenDDS开发手册---第二章（开始）3

2.1.5 数据读取者侦听器实现

    我们的侦听器类实现 由 dds
规范定义的 DDS::DataReaderListener 接口，DataReaderListener 包裹在一个 DCPS::LocalObject

解析 _narrow 和 _ptr_type 等含糊继承的成员。接口定义了一定数量的操作, 我们必须实现, 其中每一个被调用, 以通知我们不同的事件。OpenDDS:: dcp:: DataReaderListener 定义操作OpenDDS 的特殊需要, 如断开和重新连接事件更新。这里是

接口定义:

module DDS {

local interface DataReaderListener : Listener {

void on_requested_deadline_missed(in DataReader reader,

in RequestedDeadlineMissedStatus status);

void on_requested_incompatible_qos(in DataReader reader,

in RequestedIncompatibleQosStatus status);

void on_sample_rejected(in DataReader reader,

in SampleRejectedStatus status);

void on_liveliness_changed(in DataReader reader,

in LivelinessChangedStatus status);

void on_data_available(in DataReader reader);

void on_subscription_matched(in DataReader reader,

in SubscriptionMatchedStatus status);

void on_sample_lost(in DataReader reader, in SampleLostStatus status);

};

};

    我们的示例侦听器类用简单的打印将这些侦听器操作的大部分操作，此示例真正需要的唯一操作是on_data_available () 和它是这个类的唯一成员函数, 我们需要探索

void DataReaderListenerImpl::on_data_available(DDS::DataReader_ptr reader)

{

++num_reads_;

try {

Messenger::MessageDataReader_var reader_i =

Messenger::MessageDataReader::_narrow(reader);

if (!reader_i) {

std::cerr << "read: _narrow failed." << std::endl;

return;

}

   上面的代码将泛型数据读取器传递到侦听器的类型特定MessageDataReader 接口。下面的代码从消息中获取下一个示例读者.如果获取成功并返回有效数据, 我们将打印出每条消息的字段

Messenger::Message message;

DDS::SampleInfo si ;

DDS::ReturnCode_t status = reader_i->take_next_sample(message, si) ;

if (status == DDS::RETCODE_OK) {

if (si.valid_data == 1) {

std::cout << "Message: subject = " << message.subject.in() << std::endl

<< " subject_id = " << message.subject_id << std::endl

<< " from = " << message.from.in() << std::endl

<< " count = " << message.count << std::endl

<< " text = " << message.text.in() << std::endl;

}

else if (si.instance_state == DDS::NOT_ALIVE_DISPOSED_INSTANCE_STATE)

{

std::cout << "instance is disposed" << std::endl;

}

else if (si.instance_state == DDS::NOT_ALIVE_NO_WRITERS_INSTANCE_STATE)

{

std::cout << "instance is unregistered" << std::endl;

}

else

{

std::cerr << "ERROR: received unknown instance state "

<< si.instance_state << std::endl;

}

} else if (status == DDS::RETCODE_NO_DATA) {

cerr << "ERROR: reader received DDS::RETCODE_NO_DATA!" << std::endl;

} else {

cerr << "ERROR: read Message: Error: " << status << std::endl;

}

   注意示例读取可能包含无效数据。valid_data 标志指示示例具有有效数据。有两个示例传递给侦听器回调的无效数据

通知目的。一个是处理通知, 它是在DataWriter 显示地调用 dispose()。另一个是未注册的通知, 这是当 DataWriter 显式调用unregister() 时收到。处置通知将实例状态设置为 NOT_ALIVE_DISPOSED_INSTANCE_STATE通过将实例状态设置为
NOT_ALIVE_NO_WRITERS_INSTANCE_STATE
    如果有其他示例可用, 服务将再次调用此函数。然而, 阅读一次对单个样本的值并不是处理传入数据的最有效方法。数据读取器接口为处理数据提供了多种不同的选项。我们在2.2 节中讨论这些操作中的一些有效的方式。

2.1.6 在OpenDDS客户端中清理资源

    在发布服务器和订阅服务器上完成后, 我们可以使用以下代码来清除

OpenDDS 相关对象:

participant->delete_contained_entities();

dpf->delete_participant(participant);

TheServiceParticipant->shutdown ();

    域参与者的 delete_contained_entities () 操作删除所有主题,订阅者和与该参与者一起创建的发布服务器。一旦完成, 我们就可以使用域参与者工厂删除我们的域参与者。

    由于在 dds 中发布和订阅数据是分离的, 因此数据不保证全部分发， 如果发布被断开 (关闭), 则保证交付已被订阅收到。如果应用程序要求接收所有已发布的数据, 则 wait_for_acknowledgements () 操作可用于允许发布者等待所有已写的数据被收到。数据读取器必须为可靠性 qos (默认值) 提供可靠的设置, 以便wait_for_acknowledgements () 工作。此操作在单个 datawriters 上调用并包含一个超时值, 以限制等待的时间。下面的代码阐释了使用 wait_for_acknowledgements
() 可阻止长达15秒的等待订阅者来告知已经收到所有的数据:

DDS::Duration_t shutdown_delay = {15, 0};

DDS::ReturnCode_t result;

result = writer->wait_for_acknowledgments(shutdown_delay);

if( result != DDS::RETCODE_OK) {

std::cerr << "Failed while waiting for acknowledgment of "

<< "data being received by subscriptions, some data "

<< "may not have been delivered." << std::endl;

}