RTI-DDS代码分析使用介绍

DDS(Data Distribution Service数据分发服务)是对象管理组织OMG的有关分布式实时系统中数据发布的规范。
DDS规范采用了发布/订阅体系结构，但对实时性要求提供更好的支持。DDS是以数据为中心的发布/订阅通信模型。

以下工程基于rti_connext_dds-7.2.0
hello_world.idl定义的HelloWorld结构体如下

使用RTI Code Generator（rtiddsgen）生成对应工程。

生成的工程目录如下

用VS2017打开工程
重点关注 hello_world_publisher.cxx和 hello_world_subscriber.cxx两个文件

Publisher

publisher实现的是发布，subscriber实现的是订阅。
在hello_world_publisher.cxx中

    // 创建一个HelloWorld类型以HelloWorld Topic命名的Topic
    dds::topic::Topic<HelloWorld> topic(participant, "HelloWorld Topic");

    dds::pub::Publisher publisher(participant);

    // DataWriter将要在"HelloWorld Topic"中写入数据
    dds::pub::DataWriter<HelloWorld> writer(publisher, topic);
1
2
3
4
5
6
7

定义HelloWorld类型的sample，sample.msg()定义sample输出的内容
通过write函数写入数据

    HelloWorld sample;
    for (unsigned int count = 0;
         !shutdown_requested && count < sample_count;
         count++) {

		sample.msg("Hello world! " + std::to_string(count));

        std::cout << "Writing HelloWorld, count " << count << std::endl;

        writer.write(sample);

        rti::util::sleep(dds::core::Duration(4));
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

Subscriber

类似的，定义Subscriber

    // 创建一个HelloWorld类型以HelloWorld Topic命名的Topic
    dds::topic::Topic<HelloWorld> topic(participant, "HelloWorld Topic");

    dds::sub::Subscriber subscriber(participant);

    // DataReader将要读取topic数据
    dds::sub::DataReader<HelloWorld> reader(subscriber, topic);
1
2
3
4
5
6
7

创建状态条件，满足条件才读取数据

    // 创建条件
    dds::core::cond::StatusCondition status_condition(reader);

    status_condition.enabled_statuses(
            dds::core::status::StatusMask::data_available());

    // 条件触发后关联句柄
    unsigned int samples_read = 0;
    status_condition.extensions().handler([&reader, &samples_read]() {
        samples_read += process_data(reader);
    });
1
2
3
4
5
6
7
8
9
10
11

	dds::core::cond::WaitSet waitset;
    waitset += status_condition;

    while (!shutdown_requested && samples_read < sample_count) {
        // 条件激活
        std::cout << "HelloWorld subscriber sleeping for 4 sec..."
                  << std::endl;

        waitset.dispatch(dds::core::Duration(4));  // Wait up to 4s each time
    }
1
2
3
4
5
6
7
8
9
10