openDDS C++ 示例解析

现在开始

使用DCPS

本节重点介绍一个使用DCPS将数据从单个publisher 进程分发到单个subscriber 进程的示例应用程序。它基于一个简单的Messager应用程序，其中一个publisher 发布消息，一个subscriber 订阅消息。我们使用默认的QoS属性和默认的TCP/IP传输。此示例的完整源代码可以在DevGuideExamples/DCPS/Messenger/目录下找到。其他DDS和DCPS功能将在后面的章节中讨论。

使用IDL定义数据类型

在本例中，将使用OMG接口定义语言（IDL）定义topic的数据类型。有关如何不使用IDL来定义topic的OpenDDS应用程序的详细信息，请参阅DynamicDataWriters和DynamicDataReaders。

标识Topic

DDS使用的每种数据类型都是使用OMG接口定义语言（IDL）定义的。OpenDDS使用IDL注释来标识它传输和处理的数据类型。这些数据类型由TAO IDL编译器和OpenDDS IDL编译器处理，以生成使用OpenDDS传输这些类型的数据所需的代码。以下是定义Message数据类型的IDL文件：

module Messenger {

  @topic
  struct Message {
    string from;
    string subject;
    @key long subject_id;
    string text;
    long count;
  };
};
1
2
3
4
5
6
7
8
9
10
11

@topic注释标记了一个可以用作topic类型的数据类型。这必须是一个结构体或联合体。这个结构体或联合体可以包含基本类型（short, long, float等）、枚举、字符串、序列、数组、结构体和联合体。有关将IDL用于OpenDDS主题类型的更多详细信息，请参阅IDL Compliance。上面的IDL定义了Messenger模块中的Message结构，以便在本例中使用。

Keys

@key注释标识用作此topic类型的键字段。一个topic类型可以有0个或多个关键字字段。这些键用于标识topic中的不同DDS实例。键可以是标量类型、包含键字段的结构体或联合体，也可以是结构体的数组。

使用单独的@key注释指定多个键。在上面的示例中，我们将Messenger::Message的subject_id成员标识为键。使用唯一subject_id值发布的每个样本将被定义为属于同一主题中的不同DDS实例。由于我们使用默认的QoS策略，因此具有相同subject_id值的后续实例将被视为该DDS实例的替换值。
@key可以应用于以下类型的结构体字段：

• 任何原语，如布尔、整数、字符和字符串。
• 具有已定义键或一组键的其他结构体。例如：

struct StructA {
  @key long key;
};

struct StructB {
  @key StructA main_info;
  long other_info;
};

@topic
struct StructC {
  @key StructA keya; // keya.key 是key
  @key StructB keyb; // keyb.main_info.key 是另一个key
  DDS::OctetSeq data;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

在本例中，从topic类型上标记的键到用作键的原始数据类型的每种类型都用@key进行注释。但这并不是严格必要的，正如下一节所示。

• 没有任何已定义键的其他结构体。在下面的示例中，它暗示InnerStruct中的所有字段都是键。

struct InnerStruct {
  long a;
  short b;
  char c;
};

@topic
struct OuterStruct {
  @key InnerStruct value;
  // value.a, value.b, and value.c 全都是key
};
1
2
3
4
5
6
7
8
9
10
11

如果结构体中没有任何字段用@key或@key(TRUE)标记，则当该结构体在另一个结构体中使用并标记为键时，该结构体中的所有字段都被假定为键。标有@key(FALSE)的字段总是被排除在键之外，例如本例中：

struct InnerStruct {
  long a;
  short b;
  @key(FALSE) char c;
};

@topic
struct OuterStruct {
  @key InnerStruct value;
  // 那么现在就只有 value.a 和 value.b 是 keys
};
1
2
3
4
5
6
7
8
9
10
11

• 如果联合体的discriminator被标记为键，则联合体也可以用作键。在下一节中有一个键控联合主题类型的示例，但请记住，用作键的联合体不一定是topic类型。
• @key不能应用于序列，即使基类型在数组中有效。当应用于数组时@key使数组中的每个元素都成为键的一部分,不能应用于单个数组元素。

联合体Topic类型

联合体可以用作topic类型。下面是一个示例：

enum TypeKind {
  STRING_TYPE,
  LONG_TYPE,
  FLOAT_TYPE
};

@topic
union MyUnionType switch (@key TypeKind) {
case STRING_TYPE:
  string string_value;
case LONG_TYPE:
  long long_value;
case FLOAT_TYPE:
  float float_value;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

联合体类似于结构体，但只有联合体的discriminator可以是键，因此discriminator的值是带键联合体topic的潜在DDS实例集。通过将@key放在discriminator类型之前，可以为联合体topic类型指定键，如上面的示例所示。与结构体一样，也可以没有键字段，在这种情况下，@key将被省略，并且只有一个DDS实例。

Topic类型与嵌套类型

除了@topic之外，OpenDDS可以使用的IDL类型集也可以用@nested和@default_nested来控制。被 "嵌套 "的类型与topic类型相反；它们不能被用于topic的顶层类型，但它们可以被嵌套在顶层类型里面（任何级别的嵌套）。在OpenDDS中，所有的类型都是默认嵌套的，以减少类型支持所产生的代码，但是有很多方法可以改变这一点：

• 该类型可以用@topic（参见上文“标识Topic”）来注释，或者用@nested(FALSE)来注释，它等同于@topic。
• 封闭模块可以用 @default_nested(FALSE)注释。
• opendds_idl的全局默认值可以通过添加--no-default-nested来改变，在这种情况下，就像所有有效的类型都用@topic标记一样。如果需要与其他DDS的实现IDL兼容性，那么可以通过构建系统来完成：
• 使用MPC时，添加dcps_ts_flags += --no-default-nested。
• 使用CMake时，可以通过将OPENDDS_DEFAULT_NESTED全局变量设置为FALSE或将--no DEFAULT NESTED添加到OPENDDS.TARGET_SOURCES的OPENDDS_IDL_OPTIONS参数中来实现。有关将OpenDDS与cmake一起使用的详细信息，请参阅源代码中的$DDS_ROOT/docs/cmake.md。

在模块默认没有嵌套的情况下，可以通过对结构体/联合体使用@nested或@nesteed(TRUE)，对模块使用@default_nested或@default_nested(TRUE)来扭转这一局面。注意：@topic注释不接受布尔参数，因此@topic(FALSE)将在OpenDDS IDL编译器中导致错误。

对IDL的处理

本节使用OMG IDL到C++的映射（“C++ classic”）作为逐步解释的一部分。OpenDDS也支持OMG IDL-to-C++11映射，详见使用IDL-to-C++11。
OpenDDS的IDL首先由TAO IDL编译器处理。

tao_idl Messenger.idl
1

此外，我们需要用OpenDDS的IDL编译器来处理IDL文件，以生成OpenDDS所需的，序列化和键以及支撑代码以及对数据readers 和writers类型的支撑代码，来对Message进行编排和解编。这个IDL编译器位于bin中，每处理一个IDL文件就会生成三个文件。这三个文件都以原始的IDL文件名开头，显示如下：

• <filename>TypeSupport.idl
• <filename>TypeSupportImpl.h
• <filename>TypeSupportImpl.cpp

例如，按如下所示运行opendds_idl

opendds_idl Messenger.idl
1

会生成MessengerTypeSupport.idl，MessengerTypeSupportImpl.h和MessengerTypeSupportImpl.cpp。IDL文件包含MessageTypeSupport、MessageDataWriter和MessageDataReader接口定义。这些都是特定类型的DDS接口，我们以后会用这些接口在域中注册我们的数据类型，发布该数据类型的样本，并接收发布的样本。实现文件包含这些接口的实现。生成的IDL文件本身应该用TAO IDL编译器进行编译，以生成存根和骨架（stubs and skeletons）。这些文件和实现文件应该与你的使用消息类型的OpenDDS应用程序链接。OpenDDS IDL编译器有许多选项，可以使生成的代码专业化。这些选项在opendds_idl中描述。

通常情况下，你不会像上面那样直接调用TAO或OpenDDS的IDL编译器，而是让你的构建系统为你做这件事。对于使用OpenDDS的项目，支持两种不同的构建系统：

• MPC，“Make Project Creator”，用于构建OpenDDS本身及其中的大多数测试和例子。
• CMake，一个业界普遍使用的构建系统(cmake.org)

即使你最终会使用一些定制的构建系统，而不是上面列出的两个系统之一，也要先用支持的构建系统之一构建一个OpenDDS应用程序的例子，然后将代码生成器的命令行、编译器选项等迁移到定制的构建系统。

本节的其余部分将假设MPC。关于使用CMake的更多细节，请参见OpenDDS资源库中的附带文档：docs/cmake.md

使用MPC时，通过继承dcps基础项目，代码生成过程被简化。下面是publisher和subscriber共有的MPC文件部分

project(*idl): dcps {
  // This project ensures the common components get built first.

  TypeSupport_Files {
    Messenger.idl
  }
  custom_only = 1
}
1
2
3
4
5
6
7
8

dcps父项目添加了类型支持的自定义构建规则。上面的TypeSupport_Files部分告诉MPC使用OpenDDS IDL编译器从Messenger.idl生成消息类型支持文件。这里是publisher部分：

project(*Publisher): dcpsexe_with_tcp {
  exename = publisher
  after  += *idl

  TypeSupport_Files {
    Messenger.idl
  }

  Source_Files {
    Publisher.cpp
  }
}
1
2
3
4
5
6
7
8
9
10
11
12

DCPS库中的dcpsexe_with_tcp链接。

为了完整起见，这里是MPC文件的subscriber部分：

project(*Subscriber): dcpsexe_with_tcp {

  exename = subscriber
  after  += *idl

  TypeSupport_Files {
    Messenger.idl
  }

  Source_Files {
    Subscriber.cpp
    DataReaderListenerImpl.cpp
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

一个简单的信息Publisher

在这一节中，我们描述了建立一个简单的OpenDDS发布流程的步骤。代码被分成逻辑部分，并在我们介绍每一部分时进行解释。我们省略了代码中一些无趣的部分（如#include指令、错误处理和跨进程同步）。这个示例发布器的完整源代码可以在DevGuideExamples/DCPS/Messenger/中的Publisher.cpp和Writer.cpp文件中找到。

初始化参与者（Participant）

main()的第一部分将当前进程初始化为一个OpenDDS的参与者。

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;
    }
    // ...
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

TheParticipantFactoryWithArgs宏被定义在Service_Participant.h中，它用命令行参数初始化Domain Participant Factory。这些命令行参数被用来初始化OpenDDS服务使用的ORB以及服务本身。这使得我们可以在命令行上传递ORB_init()选项，以及-DCPS*形式的OpenDDS配置选项。可用的OpenDDS选项在运行时配置中得到了充分的描述。

用户可以使用(0x0 ~ 0x7FFFFFFF)范围内的ID定义任意数量的域。所有其他的值都保留给实现的内部使用。

然后，返回的域参与者对象引用被用来注册我们的消息数据类型。

注册数据类型并创建一个topic

首先，我们创建一个MessageTypeSupportImpl对象，然后使用register_type()操作用一个类型名来注册这个类型。在这个例子中，我们用一个nil字符串的类型名注册类型，这使得MessageTypeSupport接口存储库的标识符被用作类型名。也可以使用一个特定的类型名，如 “Message”。

Messenger::MessageTypeSupport_var mts =
  new Messenger::MessageTypeSupportImpl();
if (DDS::RETCODE_OK != mts->register_type(participant, "")) {
  std::cerr << "register_type failed." << std::endl;
  return 1;
}
1
2
3
4
5
6

接下来，我们从类型支持对象中获得注册的类型名称，并通过在create_topic()操作中将类型名称传递给参与者来创建topic。

CORBA::String_var type_name = mts->get_type_name ();

DDS::Topic_var topic =
  participant->create_topic ("Movie Discussion List",
                             type_name,
                             TOPIC_QOS_DEFAULT,
                             0,  // No listener required
                             OpenDDS::DCPS::DEFAULT_STATUS_MASK);
if (!topic) {
  std::cerr << "create_topic failed." << std::endl;
  return 1;
}
1
2
3
4
5
6
7
8
9
10
11
12

我们创建了一个名为 "Movie Discussion List "的topic，它具有注册类型和默认的QoS策略。

创建一个Publisher

现在，我们准备用默认的发布器QoS创建发布器。

DDS::Publisher_var pub =
  participant->create_publisher(PUBLISHER_QOS_DEFAULT,
                                0,  // No listener required
                                OpenDDS::DCPS::DEFAULT_STATUS_MASK);
if (!pub) {
  std::cerr << "create_publisher failed." << std::endl;
  return 1;
}
1
2
3
4
5
6
7
8

创建一个DataWriter并等待Subscriber

有了Publisher，我们创建DataWriter。

// 创建一个DataWriter
  DDS::DataWriter_var writer =
    pub->create_datawriter(topic,
                           DATAWRITER_QOS_DEFAULT,
                           0,  // No listener required
                           OpenDDS::DCPS::DEFAULT_STATUS_MASK);
  if (!writer) {
    std::cerr << "create_datawriter failed." << std::endl;
    return 1;
  }
1
2
3
4
5
6
7
8
9
10

当我们创建DataWriter时，我们传递topic对象引用、默认的QoS策略和一个空的监听器引用。现在我们把DataWriter引用缩小（narrow）到MessageDataWriter对象引用，这样我们就可以使用特定类型的发布操作。

Messenger::MessageDataWriter_var message_writer =
     Messenger::MessageDataWriter::_narrow(writer);
1
2

这个例子的代码使用了conditions 和wait 集合，所以Publisher会等待Subscriber连接并完全初始化。在这样一个简单的例子中，不等待Subscriber可能会导致Publisher在Subscriber连接之前发布其样本。

等待Subscriber所涉及的基本步骤是：

1. 从我们创建的DataWriter中获取状态条件
2. 在conditions 中启用发布匹配
3. 创建一个wait集合
4. 将状态条件附加到wait集合上
5. 获得发布物的匹配状态
6. 如果当前的匹配数是一个或更多，将conditions 从wait 集合中分离出来并继续发布
7. 在wait集合上等待（可以用一个指定的时间段来限定）
8. 循环回到步骤5

下面是相应的代码：

// 阻断，直到Subscriber可用

// 1.从我们创建的DataWriter中获取状态条件
DDS::StatusCondition_var condition = writer->get_statuscondition();
// 2.在conditions 中启用发布匹配
condition->set_enabled_statuses(DDS::PUBLICATION_MATCHED_STATUS);
// 3.创建一个wait集合 
DDS::WaitSet_var ws = new DDS::WaitSet;
// 4.将状态条件附加到wait集合上
ws->attach_condition(condition);

while (true) {
  DDS::PublicationMatchedStatus matches;
  // 5.获得发布物的匹配状态
  if (writer->get_publication_matched_status(matches) != DDS::RETCODE_OK) {
    std::cerr << "get_publication_matched_status failed!"
              << std::endl;
    return 1;
  }
  // 6.如果当前的匹配数是一个或更多，将conditions 从wait 集合中分离出来并继续发布
  if (matches.current_count >= 1) {
    break;
  }

  DDS::ConditionSeq conditions;
  DDS::Duration_t timeout = { 60, 0 };
  // 7.在wait集合上等待（可以用一个指定的时间段来限定）
  if (ws->wait(conditions, timeout) != DDS::RETCODE_OK) {
    std::cerr << "wait failed!" << std::endl;
    return 1;
  }

}

ws->detach_condition(condition);
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

关于状态、条件和等待集的更多细节，请参见条件和监听器。

Sample Publication

信息发布是非常直接的：

// Write samples
Messenger::Message message;
message.subject_id = 99;
message.from = "Comic Book Guy";
message.subject = "Review";
message.text = "Worst. Movie. Ever.";
message.count = 0;
for (int i = 0; i < 10; ++i) {
  DDS::ReturnCode_t error = message_writer->write(message, DDS::HANDLE_NIL);
  ++message.count;
  ++message.subject_id;
  if (error != DDS::RETCODE_OK) {
    // Log or otherwise handle the error condition
    return 1;
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

对于每个循环迭代，调用write()会导致一条消息被分发到所有为我们的topic注册的连接的Subscriber那里。由于subject_id是Message的关键，每次subject_id被递增并调用write()时，就会创建一个新的实例（见上文“Topic”）。write()的第二个参数指定了我们要发布样本的实例。它应该传递由register_instance()返回的句柄或DDS::HANDLE_NIL。传递DDS::HANDLE_NIL值表明，DataWriter应该通过检查样本的键来确定实例。关于在发布过程中使用实例句柄的详情，请参见在Publisher中注册和使用实例。

设置Subscriber

Subscriber的大部分代码与我们刚刚完成探索的Publisher相同或相似。我们将快速浏览相似的部分，并参考上面的讨论以了解细节。这个样本Subscriber的完整源代码可以在DevGuideExamples/DCPS/Messenger/的Subscriber.cpp和DataReaderListener.cpp文件中找到。

初始化参与者（Participant）

在初始化服务和加入我们的域时，Subscriber的初始化与Publisher是相同的：

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;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

注册数据类型并创建一个topic

接下来，我们初始化消息类型和topic。注意，如果topic已经在这个域中以相同的数据类型和兼容的QoS被初始化了，create_topic()调用会返回一个与现有topic对应的引用。如果我们在create_topic()调用中指定的类型或QoS与现有的主题不匹配，那么调用就会失败。还有一个find_topic()操作，我们的Subscriber可以用来简单地检索一个现有的topic。

Messenger::MessageTypeSupport_var mts =
  new Messenger::MessageTypeSupportImpl();
if (DDS::RETCODE_OK != mts->register_type(participant, "")) {
  std::cerr << "Failed to register the MessageTypeSupport." << std::endl;
  return 1;
}

CORBA::String_var type_name = mts->get_type_name();

DDS::Topic_var topic =
  participant->create_topic("Movie Discussion List",
                            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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

创建一个Subscriber

接下来，我们用默认的QoS创建Subscriber。

// Create the subscriber
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;
}
1
2
3
4
5
6
7
8
9

创建一个 DataReader 和监听器

我们需要将一个监听器对象与我们创建的DataReader联系起来，这样我们就可以用它来检测数据何时可用。下面的代码构建了监听器对象。DataReaderListenerImpl类显示在下一小节。

DDS::DataReaderListener_var listener(new DataReaderListenerImpl);
1

监听器被分配到堆内存上，并分配给DataReaderListener_var对象。这种类型提供了引用计数行为，所以当最后一个引用被移除时，监听器会被自动回收。这种用法在OpenDDS应用代码中是典型的堆分配，并使应用开发者不必主动管理分配对象的寿命。

现在我们可以创建DataReader ，并将其与我们的topic、默认的QoS属性以及我们刚刚创建的监听器对象联系起来。

// Create the Datareader
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;
}
1
2
3
4
5
6
7
8
9
10

这个线程现在可以自由地执行其他应用工作。当有样本时，我们的监听器对象将在OpenDDS线程上被调用。

数据读取器监听器的实现

我们的监听器类实现了DDS::DataReaderListener接口，由DDS规范定义。DataReaderListener被包装在一个DCPS::LocalObject中，该对象解析不明确继承的成员，如_narrow和_ptr_type。该接口定义了我们必须实现的许多操作，每个操作都被调用以通知我们不同的事件。OpenDDS:：DCPS:：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);
  };
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

我们的示例监听器类用简单的打印语句将这些监听器的大部分操作存根出来。本例中唯一真正需要的操作是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;
    }
1
2
3
4
5
6
7
8
9
10
11

上面的代码将传入监听器的通用DataReader缩小到特定类型的MessageDataReader接口。下面的代码从DataReader中获取下一个样本。如果获取成功并返回有效数据，我们将打印出消息的每个字段。

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

请注意，样本读取可能包含无效数据。valid_data标志指示样本是否具有有效数据。有两个包含无效数据的样本交付给监听器回调，用于通知。一个是dispose通知，它是在DataWriter显式调用dispose()时接收的。另一个是unregistered 通知，它是在DataWriter显式调用unregister()时接收的。发送dispose通知时，实例状态设置为NOT_ALIVE_DISPOSED_INSTANCE_STATE，发送unregister通知时，示例状态设置为NOT_ALIVE_NO_WRITERS_INSTANCE_STATE。

如果有其他示例可用，服务将再次调用此函数。然而，一次读取单个样本的值并不是处理传入数据的最有效方法。DataReader接口提供了许多不同的选项，用于以更有效的方式处理数据。我们在数据处理优化中讨论了其中一些操作。

在OpenDDS客户端进行清理

在我们完成了Publisher和Subscriber的工作后，我们可以使用下面的代码来清理OpenDDS相关的对象：

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

域参与者（domain participant）的delete_contained_entities()操作会删除所有用该参与者创建的topic、subscriber和publisher。一旦完成这个操作，我们就可以使用domain participant factory来删除我们的域参与者。

由于DDS中数据的publish和subscribe是解耦的，如果在所有已publish的数据被subscriber收到之前，一个publication 被解耦（关闭），则不能保证数据的交付。如果应用程序要求收到所有发布的数据，那么可以使用wait_for_acknowledgments()操作来允许publication 等待，直到所有写入的数据都被收到。dataReader必须有RELIABILITYQoS的RELIABLE设置（这是默认的），以便wait_for_acknowledgments()能够工作。这个操作是在单个dataWriter上调用的，包括一个超时值来约束等待的时间。下面的代码说明了如何使用wait_for_acknowledgments()来阻塞长达15秒的时间，以等待subscriber确认收到所有写入的数据：

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;
}
1
2
3
4
5
6
7
8

运行示例

现在我们已经准备好运行我们的简单例子了。在自己的窗口中运行这些命令，应该可以让你最容易理解输出。

首先，我们将启动一个DCPSInfoRepo服务，这样我们的publisher和subscriber就可以互相找到。

如果将环境配置为使用RTPS发现来使用对等发现，则不需要执行此步骤。

DCPSInfoRepo的可执行文件在bin/DCPSInfoRepo中找到。当我们启动DCPSInfoRepo时，我们需要确保publisher和subscriber的应用程序进程也能找到启动的DCPSInfoRepo。这个信息可以通过以下三种方式之一提供：

1. 命令行上的参数
2. 生成并放在共享文件中供应用程序使用
3. 放在配置文件中供其他进程使用的参数。

对于我们这里的简单例子，将使用第二种方法，将DCPSInfoRepo的位置属性生成到一个文件中，这样我们简单的publisher和subscriber就可以读取它并连接到它：

%DDS_ROOT%\bin\DCPSInfoRepo -o simple.ior
1

-o参数指示DCPSInfoRepo生成其到文件simple.ior的连接信息，供publisher和subscriber使用。在单独的窗口中，导航到包含simple.ior文件的同一目录，并通过键入以下内容来启动我们示例中的subscriber应用程序：

subscriber -DCPSInfoRepo file://simple.ior
1

publisher连接到DCPSInfoRepo以找到任何subscriber的位置，并开始发布消息以及将它们写到控制台。在subscriber窗口中，你现在也应该看到来自subscriber的控制台输出，该subscriber正在从topic中读取消息，演示一个简单的publish和subscribe应用程序。

在配置DDSI-RTPS发现和RTPS_UDP传输配置选项中，可以阅读有关为RTPS和其他更高级配置选项配置应用程序的更多信息。有关配置和使用DCPSInfoRepo的信息，请参阅发现配置和DCPS信息存储仓库。有关设置和使用修改应用程序行为的QoS功能的信息，请参阅服务质量。

用RTPS运行示例

之前的OpenDDS例子展示了如何使用基本的OpenDDS配置和使用DCPSInfoRepo服务的集中发现来构建和执行一个OpenDDS应用。下面将详细介绍使用RTPS进行发现并使用可互操作的传输来运行同一个例子所需要的东西。当你的OpenDDS应用需要与DDS规范的非OpenDDS实现互操作，或者你不想在OpenDDS的部署中使用集中发现时，这一点很重要。

上面的Messenger例子的编码和构建并没有因为使用RTPS而改变，所以你不需要修改或重建你的publisher和subscriber服务。这是OpenDDS架构的一个优势，因为要启用RTPS功能，这只是一个配置练习。在这个练习中，我们将使用publisher和subscriber共享的配置文件，为Messenger的例子启用RTPS。关于包括RTPS在内的所有可用传输的配置的更多细节将在运行时配置中描述。

导航到构建publisher和subscriber的目录。创建名为rtps.ini的新文本文件，并用以下内容填充它：

[common]
DCPSGlobalTransportConfig=$file
DCPSDefaultDiscovery=DEFAULT_RTPS

[transport/the_rtps_transport]
transport_type=rtps_udp
1
2
3
4
5
6

值得关注的两行是设置发现方法的一行和设置数据传输协议为RTPS的一行。

现在让我们重新运行我们的例子，启用RTPS，先启动subscriber进程，然后启动publisher开始发送数据。最好是在不同的窗口中启动它们，以便看到这两个进程分别工作。

使用-DCPSConfigFile命令行参数启动subscriber，以指向新创建的配置文件…

subscriber -DCPSConfigFile rtps.ini
1

以同样参数启动publisher：

publisher -DCPSConfigFile rtps.ini
1

由于在RTPS规范中没有集中的发现，因此存在允许发生发现的等待时间的规定。该规范将默认时间设置为30秒。当上述两个进程被启动时，可能有高达30秒的延迟，这取决于他们彼此之间的启动距离。这个时间可以在OpenDDS配置文件中调整，并在配置DDSI-RTPS发现中讨论。

由于OpenDDS的体系结构允许可插拔发现和可插拔传输，因此可以单独更改上面rtps.ini文件中调用的两个配置条目，一个使用RTPS，另一个不使用RTPS（例如，使用DCPSInfoRepo的集中发现）。在我们的示例中，将它们都设置为RTPS可以使该应用程序与其他非OpenDDS实现完全互操作。

原文链接详见：https://opendds.readthedocs.io/en/latest-release/devguide/getting_started.html#cleaning-up-in-opendds-clients