OpenDDS

基于OpenDDS应用程序开发(3)订阅端实现

连续的三篇博文演示如何基于OpenDDS开发应用程序，将数据从发布端节点发送到订阅端节点，该示例程序由一个发布者发布数据，一个订阅者订阅数据，使用默认的QoS策略和TCP/IP传输方式。

本文是第三篇，主要介绍开发一个简单的OpenDDS订阅端应用程序所涉及的步骤。省略一些不重要部分(如:#include部分和异常处理等)代码，只写出关键代码。

1、新建订阅端工程：

参考前一博文中MPC的用法，在Demo.mpc文件中增加如下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 project(*Subscriber) : dcpsexe_with_tcp {

 exename = subscriber
 after += *idl

 TypeSupport_Files {
   Demo.idl
 }

 Source_Files {
   Subscriber.cpp
   DataReaderListenerImpl.cpp
 }
}

Subscriber工程从父工程dcpsexe_with_tcp继承，这里直接使用idl工程中定义好的Demo.idl文件。

之后在Demo目录下新建三个文件：Subscriber.cpp、DataReaderListenerImpl.h、DataReaderListenerImpl.cpp，分别用来编写订阅端逻辑部分代码，并再次使用如下命令来生成Vs2008工程：

1
mwc.pl -type vc9

生成完成之后，使用Vs2008打开Demo.sln，就可以修改订阅端代码了：

2、初始化参与者：

初始化订阅端参与者代码同发布端是完全一样的，在Subscriber.cpp文件中增加如下内容：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
int main(int argc, char *argv[])
{
  try {
  
   DDS::DomainParticipantFactory_var dpf =
     TheParticipantFactoryWithArgs(argc, argv);

   DDS::DomainParticipant_var participant =
     dpf->create_participant(42, // Domain ID
                         PARTICIPANT_QOS_DEFAULT,
                         0,  // No listener required
                         OpenDDS::DCPS::DEFAULT_STATUS_MASK);
   if (!participant) {
     std::cerr << "create_participant failed." << std::endl;
     return 1 ;
   }

3、注册数据类型并创建主题：

接下来，初始化数据类型和主题：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Demo::PosTypeSupport_var mts = new Demo::PosTypeSupportImpl();
if (DDS::RETCODE_OK != mts->register_type(participant, "")) {
  std::cerr << "Failed to register the PosTypeSupport." << std::endl;
  return 1;
}

CORBA::String_var type_name = mts->get_type_name();
DDS::Topic_var topic =
  participant->create_topic("Pos Demo",
                        type_name,
                        TOPIC_QOS_DEFAULT,
                        0,   // No listener required
                        OpenDDS::DCPS::DEFAULT_STATUS_MASK);
if (!topic) {
  std::cerr << "Failed to create_topic." << std::endl;
  return 1;
}

4、创建订阅者：

调用create_subscriber()操作创建一个带有默认QoS策略的订阅者：

1
2
3
4
5
6
7
8
DDS::Subscriber_var sub =
  participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT,
                            0,  // No listener required
                            OpenDDS::DCPS::DEFAULT_STATUS_MASK);
if (!sub) {
  std::cerr << "Failed to create_subscriber." << std::endl;
  return 1;
}

5、创建数据读者及监听者：

订阅端需要给数据读者关联一个监听者，用来接收数据的到达，下面的代码定义了一个监听者对象，类DataReaderListenerImpl的实现将在下一部分介绍。

1
DDS::DataReaderListener_var listener(new DataReaderListenerImpl);

现在采用默认的QoS策略创建数据读者，并将它与主题、刚刚创建的监听者对象相关联起来：

1
2
3
4
5
6
7
8
9
DDS::DataReader_var dr =
  sub->create_datareader(topic,
                     DATAREADER_QOS_DEFAULT,
                     listener,
                     OpenDDS::DCPS::DEFAULT_STATUS_MASK);
if (!dr) {
  std::cerr << "create_datareader failed." << std::endl;
  return 1;
}

之后，主线程就可以自由的去处理其它工作了，当有数据到达时，OpenDDS会调用监听者对象的回调接口通知，只需要在DataReaderListenerImpl类的回调函数中接收需要的数据就可以了。

6、数据读者监听者实现：

监听者类继承自DDS规范的DDS::DataReaderListener接口，该接口定义了一些回调函数，每个回调函数被调用时，就是一个事件的通知，如：断开、重连等，以下是DataReaderListener接口的定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
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);

 };
};

在本例的DataReaderListenerImpl类中真正需要的实现的回调接口是on_data_available()，它也是我们需要重新派生该类的唯一成员函数：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
void DataReaderListenerImpl::on_data_available(DDS::DataReader_ptr reader) {

 num_reads_ ++;

 try{

   Demo::PosDataReader_var reader_i = Demo::PosDataReader::_narrow(reader);
   if (!reader_i) {
     std::cerr << "read: _narrow failed." << std::endl;
     return;
   }

   Demo::Pos pos;
   DDS::SampleInfo si ;
   DDS::ReturnCode_t status = reader_i->take_next_sample(pos, si) ;
   if (status == DDS::RETCODE_OK) {
   
     if (si.valid_data == 1) {
       std::cout << " Pos:pos_id = " << pos. pos_id << std::endl
                 << "  pos_x = " << pos. pos_x << std::endl
                 << "  pos_y = " << pos. pos_y << 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 Pos: " << status << std::endl;
   }

上面的代码将样本从数据读者中取出，如果成功并能返回有效数据，就打印出接收到数据的每一个字段。

每当有样本数据到达时，该函数就会被调用。

7、实体清理：

在订阅完数据以后，需要清理与OpenDDS相关联的资源：

1
2
3
participant->delete_contained_entities();
dpf->delete_participant(participant);
TheServiceParticipant->shutdown();

调用域参与者的delete_contained_entities()操作删除所有该参与者创建的主题、订阅者。一旦执行完该操作，就可以使用域参与者工厂删除域参与者了。

8、示例程序运行：

修改完以上代码并编译完成，就可以运行订阅端应用程序了，需要先运行DDS的信息仓库，开始中打开一个CMD窗口，执行如下命令：

1
%DDS_ROOT%/bin/DCPSInfoRepo  -ORBListenEndpoints  iiop://localhost:12345

再次打开一个CMD窗口，cd到Demo目录下，执行如下命令：

1
subscriber -DCPSInfoRepo corbaloc::localhost:12345/DCPSInfoRepo

至此，订阅端应用程序就开发完成并运行起来了。

有关OpenDDS的相关问题欢迎发送邮件至lyingbo@aliyun.com一起讨论