當(dāng)前位置：首頁(yè) > 编程资源 > 综合教程 >内容正文

综合教程

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

發(fā)布時(shí)間：2024/8/26 综合教程 29 生活家

生活随笔收集整理的這篇文章主要介紹了基于OpenDDS应用程序开发(3)订阅端实现小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

連續(xù)的三篇博文演示如何基于OpenDDS開(kāi)發(fā)應(yīng)用程序，將數(shù)據(jù)從發(fā)布端節(jié)點(diǎn)發(fā)送到訂閱端節(jié)點(diǎn)，該示例程序由一個(gè)發(fā)布者發(fā)布數(shù)據(jù)，一個(gè)訂閱者訂閱數(shù)據(jù)，使用默認(rèn)的QoS策略和TCP/IP傳輸方式。

本文是第三篇，主要介紹開(kāi)發(fā)一個(gè)簡(jiǎn)單的OpenDDS訂閱端應(yīng)用程序所涉及的步驟。省略一些不重要部分(如:#include部分和異常處理等)代碼，只寫(xiě)出關(guān)鍵代碼。

1、新建訂閱端工程：

參考前一博文中MPC的用法，在Demo.mpc文件中增加如下內(nèi)容：

 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目錄下新建三個(gè)文件：Subscriber.cpp、DataReaderListenerImpl.h、DataReaderListenerImpl.cpp，分別用來(lái)編寫(xiě)訂閱端邏輯部分代碼，并再次使用如下命令來(lái)生成Vs2008工程：

1	mwc.pl -type vc9

生成完成之后，使用Vs2008打開(kāi)Demo.sln，就可以修改訂閱端代碼了：

2、初始化參與者：

初始化訂閱端參與者代碼同發(fā)布端是完全一樣的，在Subscriber.cpp文件中增加如下內(nèi)容：

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、注冊(cè)數(shù)據(jù)類(lèi)型并創(chuàng)建主題：

接下來(lái)，初始化數(shù)據(jù)類(lèi)型和主題：

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、創(chuàng)建訂閱者：

調(diào)用create_subscriber()操作創(chuàng)建一個(gè)帶有默認(rèn)QoS策略的訂閱者：

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、創(chuàng)建數(shù)據(jù)讀者及監(jiān)聽(tīng)者：

訂閱端需要給數(shù)據(jù)讀者關(guān)聯(lián)一個(gè)監(jiān)聽(tīng)者，用來(lái)接收數(shù)據(jù)的到達(dá)，下面的代碼定義了一個(gè)監(jiān)聽(tīng)者對(duì)象，類(lèi)DataReaderListenerImpl的實(shí)現(xiàn)將在下一部分介紹。

1	DDS::DataReaderListener_var listener(new DataReaderListenerImpl);

現(xiàn)在采用默認(rèn)的QoS策略創(chuàng)建數(shù)據(jù)讀者，并將它與主題、剛剛創(chuàng)建的監(jiān)聽(tīng)者對(duì)象相關(guān)聯(lián)起來(lái)：

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;
大專(zhuān)欄%20基于OpenDDS應(yīng)用程序開(kāi)發(fā)(3)訂閱端實(shí)現(xiàn)ne">  return 1;
}

之后，主線程就可以自由的去處理其它工作了，當(dāng)有數(shù)據(jù)到達(dá)時(shí)，OpenDDS會(huì)調(diào)用監(jiān)聽(tīng)者對(duì)象的回調(diào)接口通知，只需要在DataReaderListenerImpl類(lèi)的回調(diào)函數(shù)中接收需要的數(shù)據(jù)就可以了。

6、數(shù)據(jù)讀者監(jiān)聽(tīng)者實(shí)現(xiàn)：

監(jiān)聽(tīng)者類(lèi)繼承自DDS規(guī)范的DDS::DataReaderListener接口，該接口定義了一些回調(diào)函數(shù)，每個(gè)回調(diào)函數(shù)被調(diào)用時(shí)，就是一個(gè)事件的通知，如：斷開(kāi)、重連等，以下是DataReaderListener接口的定義：

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類(lèi)中真正需要的實(shí)現(xiàn)的回調(diào)接口是on_data_available()，它也是我們需要重新派生該類(lèi)的唯一成員函數(shù)：

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;
   }

上面的代碼將樣本從數(shù)據(jù)讀者中取出，如果成功并能返回有效數(shù)據(jù)，就打印出接收到數(shù)據(jù)的每一個(gè)字段。

每當(dāng)有樣本數(shù)據(jù)到達(dá)時(shí)，該函數(shù)就會(huì)被調(diào)用。

7、實(shí)體清理：

在訂閱完數(shù)據(jù)以后，需要清理與OpenDDS相關(guān)聯(lián)的資源：

1
2
3

participant->delete_contained_entities();
dpf->delete_participant(participant);
TheServiceParticipant->shutdown();

調(diào)用域參與者的delete_contained_entities()操作刪除所有該參與者創(chuàng)建的主題、訂閱者。一旦執(zhí)行完該操作，就可以使用域參與者工廠刪除域參與者了。

8、示例程序運(yùn)行：

修改完以上代碼并編譯完成，就可以運(yùn)行訂閱端應(yīng)用程序了，需要先運(yùn)行DDS的信息倉(cāng)庫(kù)，開(kāi)始中打開(kāi)一個(gè)CMD窗口，執(zhí)行如下命令：

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

再次打開(kāi)一個(gè)CMD窗口，cd到Demo目錄下，執(zhí)行如下命令：

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

至此，訂閱端應(yīng)用程序就開(kāi)發(fā)完成并運(yùn)行起來(lái)了。

有關(guān)OpenDDS的相關(guān)問(wèn)題歡迎發(fā)送郵件至lyingbo@aliyun.com一起討論

總結(jié)

以上是生活随笔為你收集整理的基于OpenDDS应用程序开发(3)订阅端实现的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇：吃瓜群众的由来（吃瓜群众什么意思）
下一篇：中考必胜藏头诗