Java调用OpenDDS(2)-理解OpenDDS自带的Messager示例

OpenDDS安装好之后，下一步就是利用OpenDDS来开发通信项目了。不过在项目中应用OpenDDS之前，先消化一下OpenDDS安装包中自带的示例项目messenger，通过阅读messenger的源代码来熟悉一下OpenDDS提供的用来开发Java项目的类。

提纲
1、准备工作
2、发送消息：TestPublisher
3、接收消息：TestSubscriber & DataReaderListenerImpl
4、在IDEA中运行示例

1、准备工作

在开始看OpenDDS的示例项目之前，需要安装好OpenDDS，并且在编译OpenDDS的时候要开启Java支持，这是前提条件。

第一步，在IDEA创建空的maven项目ddstest，用这个项目来看OpenDDS自带的messenger项目的示例代码。因为Idea无法直接打开messenger项目，所以才要创建一个ddstest新项目，把messenger的源码拷贝到ddstest中去边运行边看。

第二步，ddstest项目中引入4个必要的jar包。具体方法：File - Project Structure -Libraries - 点击"+"号 - Java ，然后定位到 %DDS_ROOT%\lib，选中3个jar包：i2jrt.jar、OpenDDS_DCPS.jar、tao_java.jar，定位到%DDS_ROOT%\java\tests\messenger下\messenger_idl下，选中messenger_idl_test.jar。
这样以来项目建好了，所需要的的4个jar包也都加入到项目的libraries中了。详情见下图：

注：由于OpenDDS需要CORBA，而Java 9开始移除了CORBA，所以请在Java 8环境下开发。

然后去%DDS_ROOT%\java\tests\messenger下，将publisher和subscriber两个目录下的TestPublisher、TestSubscriber、DataReaderListenerImpl三个java文件复制到ddstest项目的src目录下。

经过以上2步，项目结构如上图所示，源代码都在src下，jar都在external libraries下面。

2、利用TestPublisher发送消息

DomainParticipantFactory dpf = TheParticipantFactory.WithArgs(new StringSeqHolder(args));
DomainParticipant dp = dpf.create_participant(4, PARTICIPANT_QOS_DEFAULT.get(), null, DEFAULT_STATUS_MASK.value);

第一行用来创建域参与者工厂，需要读取参数（实际上就是读取repo.ior文件）
第二行是创建具体的域参与者，create_participant方法的定义（实际在DomainParticipantFactoryImpl.cpp文件中）为：

DDS::DomainParticipant_ptr DomainParticipantFactoryImpl::create_participant(DDS::DomainId_t domainId,const DDS::DomainParticipantQos & qos,DDS::DomainParticipantListener_ptr a_listener,DDS::StatusMask mask)
{DDS::DomainParticipantQos par_qos = qos;if (par_qos == PARTICIPANT_QOS_DEFAULT) {get_default_participant_qos(par_qos);}if (!Qos_Helper::valid(par_qos)) {ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%t) ERROR: ")ACE_TEXT("DomainParticipantFactoryImpl::create_participant, ")ACE_TEXT("invalid qos.\n")));return DDS::DomainParticipant::_nil();}if (!Qos_Helper::consistent(par_qos)) {ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%t) ERROR: ")ACE_TEXT("DomainParticipantFactoryImpl::create_participant, ")ACE_TEXT("inconsistent qos.\n")));return DDS::DomainParticipant::_nil();}RcHandle<DomainParticipantImpl> dp =make_rch<DomainParticipantImpl>(this, domainId, par_qos, a_listener, mask);if (qos_.entity_factory.autoenable_created_entities) {if (dp->enable() != DDS::RETCODE_OK) {ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%t) ERROR: ")ACE_TEXT("DomainParticipantFactoryImpl::create_participant, ")ACE_TEXT("unable to enable DomainParticipant.\n")));return DDS::DomainParticipant::_nil();}}ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,tao_mon,this->participants_protector_,DDS::DomainParticipant::_nil());participants_[domainId].insert(dp);return dp._retn();
}

主要就是进行Qos的验证，然后创建DomainParticipantImpl对象，调用其enable方法以激活，然后添加到参与者组中。参与者组实质是个Map，以域ID作为key。

接下来是创建主题：

MessageTypeSupportImpl servant = new MessageTypeSupportImpl();
Topic top = dp.create_topic("Movie Discussion List",servant.get_type_name(),TOPIC_QOS_DEFAULT.get(),null,DEFAULT_STATUS_MASK.value);

需要注意的是，OpenDDS的主题和发送的消息，必须在发送时就全部确定，不能用Scanner接受用户输入然后动态创建

create_topic方法实际调用的是create_topic_i方法，添加了一个topic_mask参数（值为0），整个方法有150+行，仅挑些重点：

首先还是Qos检查，略过。

接下来是根据主题名检查主题有没有创建过（假如开启了主题过滤，并且不允许重复主题，则直接退出）：

TopicMap::mapped_type* entry = 0;bool found = false;{ACE_GUARD_RETURN(ACE_Recursive_Thread_Mutex,tao_mon,this->topics_protector_,DDS::Topic::_nil());#if !defined(OPENDDS_NO_CONTENT_FILTERED_TOPIC) || !defined(OPENDDS_NO_MULTI_TOPIC)if (topic_descrs_.count(topic_name)) {if (DCPS_debug_level > 3) {…… //一些报错信息}return 0;}
#endifif (Util::find(topics_, topic_name, entry) == 0) {found = true;}}

假如同名主题已经创建过了，则进行type和qos的比较，如果有一个不符合，则返回空，创建失败，如果都符合，说明主题已经存在，就直接返回主题对象

假如同名主题不存在，则创建新主题：

首先要检查欲创建的主题类型有没有注册过：

    if (0 == topic_mask) {// creating a topic with compile time typetype_support = Registered_Data_Types->lookup(this, type_name);if (CORBA::is_nil(type_support)) {if (DCPS_debug_level >= 1) {…… //报错}return DDS::Topic::_nil();}has_keys = type_support->has_dcps_key();}

接下来就是对topic信息的检查，以及初始化Topic对象的过程，如果设置了TopicListener的话，创建完毕后会进行回调

接下来是创建Publisher对象；

Publisher pub = dp.create_publisher(PUBLISHER_QOS_DEFAULT.get(), null, DEFAULT_STATUS_MASK.value);

其C++代码流程和create_participant类似，不同的是，保存Publisher对象的是个集合而非Map，还有Publisher的ID是用一个内置的ID生成器自动产生的

接下来是创建DataWriter对象，为之配置了非常复杂的Qos策略，并等待Publisher匹配成功

到这里准备工作才刚刚结束，下面开始正式的数据发送

        MessageDataWriter mdw = MessageDataWriterHelper.narrow(dw);Message msg = new Message();msg.subject_id = 99;int handle = mdw.register_instance(msg);msg.from = "OpenDDS-Java";msg.subject = "Review";msg.text = "Worst. Movie. Ever.";msg.count = 0;int ret = RETCODE_TIMEOUT.value;for (; msg.count < N_MSGS; ++msg.count) {while ((ret = mdw.write(msg, handle)) == RETCODE_TIMEOUT.value) {}if (ret != RETCODE_OK.value) {System.err.println("ERROR " + msg.count +" write() returned " + ret);}try {Thread.sleep(100);} catch(InterruptedException ie) {}}

MessageDataWriter是messenger_idl_test包含的DataWriter子类，实际执行Message发送任务，Message的格式由Messenger.idl定义。for循环一共发送40条消息。

假如启动时加了-w参数，则会等待响应，否则等待1秒后直接进行清理、退出

3、接收消息：TestSubscriber & DataReaderListenerImpl

TestSubscriber的整个流程和TestPublisher几乎完全一致，不同的是，创建DataReader时，会将DataReaderListenerImpl实例传入。

当Publisher向域中输入消息后，就会触发DATA_AVAILABLE事件，通知DataReaderListenerImpl处理

    public synchronized void on_data_available(DataReader reader) {initialize_counts();MessageDataReader mdr = MessageDataReaderHelper.narrow(reader);if (mdr == null) {System.err.println("ERROR: read: narrow failed.");return;}MessageHolder mh = new MessageHolder(new Message());SampleInfoHolder sih = new SampleInfoHolder(new SampleInfo(0, 0, 0,new Time_t(), 0, 0, 0, 0, 0, 0, 0, false, 0));int status = mdr.take_next_sample(mh, sih);if (status == RETCODE_OK.value) {System.out.println("SampleInfo.sample_rank = "+ sih.value.sample_rank);System.out.println("SampleInfo.instance_state = "+ sih.value.instance_state);if (sih.value.valid_data) {String prefix = "";boolean invalid_count = false;if (mh.value.count < 0 || mh.value.count >= counts.size()) {invalid_count = true;}else {if (counts.get(mh.value.count) == false){counts.set(mh.value.count, true);}else {prefix = "ERROR: Repeat ";}}…… //消息内容输出}…… //一些报错} else if (status == RETCODE_NO_DATA.value) {System.err.println("ERROR: reader received DDS::RETCODE_NO_DATA!");} else {System.err.println("ERROR: read Messenger.Message: Error: " + status);}}

程序使用MessageHolder存储读到的信息，每次读取一条。

4、在IDEA中运行TestPublisher、TestSubscriber

运行run_test.pl脚本时，可以看到类似：

“C:\Program Files\Java\bin\java.exe” -Xcheck:jni -ea -cp classes;E:\OpenDDS-3.13/lib/i2jrt.jar;E:\OpenDDS-3.13/lib/OpenDDS_DCPS.jar;E:\OpenDDS-3.13/java/tests/messenger/messenger_idl/messenger_idl_test.jar;publisher/classes -Dopendds.native.debug=true TestPublisher -DCPSBit 0 -DCPSConfigFile tcp.ini -r -w
的命令输出，可以参照这个配置运行条件。

1）首先配置虚拟机选项：

Run菜单 - Edit Configurations，在 vm options 一栏填写：

-ea -Dopendds.native.debug=true -Djava.library.path=E:\dds\OpenDDS-3.13/java/tests/messenger/messenger_idl;E:\OpenDDS-3.13\lib

请按照自己的实际情况修改路径。由于已经配置了jar包，所以不需要再写-cp选项。

-Djava.library.path是用来配置JNI库路径，不写的话会提示找不到dll

去除 -Xcheck:jni 是为了屏蔽掉过多的JNI Warning

2）然后配置主程序参数：

在同一个窗口的 Program arguments一栏填写：

-DCPSBit 0 -DCPSConfigFile E:\dds\OpenDDS-3.13/java/tests/messenger/tcp.ini

同样按照自己的实际情况修改路径。这一行如果写在vm options中，就会提示”ERROR: Domain Participant Factory not found“，因为参数传递给了虚拟机而不是程序，导致无法创建DomainParticipantFactory

3）运行实例

按照以上配置，分别配置TestPublisher的run configuration和TestSubscriber的run configuration，配置两个，因为运行的时候，它们两个会同时运行，各自使用自己的run configuration。