REDHAWK——连接（续）

---

前言

续上文

---

一、突发 IO

对于那些需要小型且可能是非连续的数据块（或突发）以及频繁变化的元数据的应用程序，突发输入/输出（BurstIO）提供了满足这些要求的数据传输容器和接口。这个接口仅支持数据向量的传输：float, double, octet (int8/uint8), short (int16), ushort (uint16), long (int32), ulong (uint32), longlong (int64), 和 ulonglong(uint64)。与批量输入/输出（BulkIO）类似，BurstIO 提供了突发信号相关信息（SRI）和精确时间戳，但是它通过每个数据突发中的带内信息提供这些信息。由于元数据的增加的开销要求，通过将多个突发分组为单次传输，无论是通过编程还是通过可配置的策略设置，BurstIO 可以实现其最高吞吐量，以尝试最大化效率并限制延迟。

1、数据传输

突发输入/输出（BurstIO）数据传输通过 REDHAWK 资源（组件或设备）的 BurstIO 端口对象的 pushBurst() 和 pushBursts() 方法调用来实现。资源可以使用这些 push 方法将突发及其关联的元数据从一个资源传输到另一个资源，这一过程在资源的服务功能内进行。与批量输入/输出（BulkIO）相似，BurstIO 接口为突发的每个数据向量提供相同的 BULKIO::PrecisionUTCTime 时间戳。BurstIO 定义了一个新的 BurstSRI 信号相关信息（SRI）对象，使开发人员能够进一步描述信号环境和数据转换。这些字段在突发信号相关信息（SRI）中有进一步的描述。

①、输入

拥有提供端口（输入端口）的资源，使用 getBurst() 方法从端口抓取数据。此方法从输入端口的数据队列返回一个 PacketType 对象（在突发包访问器中描述），如果队列为空，则返回 null/None 值。

以下代码片段是 getBurst() 方法的一个示例。

/**Grab data from the port's getBurst method*/
burstio::BurstShortIn::PacketType *pkt;
pkt = inShortPort->getBurst( bulkio::Const::NON_BLOCKING );// check if a valid packet was returned
if ( pkt == NULL ) {return NOOP;
}// check for EOS
if ( pkt->getEOS() ) {outShortPort->pushBurst(pkt->getSequence(), pkt->getSRI(), pkt->getEOS());
}...  perform algorithm on the data: pkt->getData() ... or pkt->getSequence()

②、输出

由于 BurstIO 数据的异步性，该接口使开发者能够控制从资源中输出（输出）突发。将突发数据从一个资源推送到另一个资源的两个主要方法调用是：pushBursts() 和 pushBurst()。pushBursts() 允许直接将多个突发作为 BurstType 对象的序列直接下游发送，而 pushBurst() 提供了一个接口来排队推送单个突发，但根据突发的数量、总队列大小和发送间隔遵循策略指令。这两种方法都使用指定的路由约束和连接过滤器路由突发数据，这些可以通过以下接口控制：

// this route streams with Stream ID == "data-stream-one" to a connection
// identified as "connection-one"
shortBurstPort->addConnectionFilter("data-stream-one", "connection-one");

或者：

// update connection filter using the Component's connection property
// "myConnectionTable"
shortBurstPort->updateConnectionFilter(myConnectionTable);// this sets the stream filter to only route streams to specific connections
shortBurstPort->setRoutingMode(burstio::ROUTE_CONNECTION_STREAMS);

pushBurst() 和 pushBursts() 方法之间的主要区别在于管理数据传输的方式和时间的能力。只有使用 pushBurst() 排队的突发流量受到策略约束的控制，而对 pushBursts() 的调用则直接发送到下游连接的资源。

// this method will limit the maximum number of bursts that
// can be queued  before they are sent
shortBurstPort->setMaxBursts(size_t count);// this method will enable threshold monitoring for the amount of sample
// data that exceeds this limit before sending data downstream
shortBurstPort->setByteThreshold(size_t bytes);// this method will enable the latency time between the sending of
// available data downstream
shortBurstPort->setLatencyThreshold( long usec );

以下代码片段是一个使用 pushBurst() 方法调用将向量数据样本排队到端口的示例。

std::vector< BurstShortOut::NativeType > data;
my_transform(data);BURSTIO::BurstSRI  sri;
burstio::BurstShortOut::BurstType burst;
burst.SRI = sri;
burst.EOS = false;
burst.T = burstio::utils::now();
burst.data.length(data.size());
for(int i=0; i< data.size(); i++ ) burst.data[i] = data[i];// this queues a single burst
shortBurstPort->pushBurst( burst );// orstd::vector< BurstShortOut::NativeType > data;
my_transform(data);// this queues a single burst
shortBurstPort->pushBurst( data, sri, burstio::utils::now() );

以下代码片段是一个使用 pushBursts() 方法调用的向量数据样本的示例。此调用中的突发直接传递到下游的连接资源。

std::vector< BurstShortOut::NativeType > data;
my_transform(data);BurstShortOut::BurstSequenceType bursts;
bursts.length(1);
burstio::BurstShortOut::BurstType burst;
burst.SRI = sri;
burst.EOS = false;
burst.T = burstio::utils::now();
burst.data.length(data.size());
for(int i=0; i< data.size(); i++ ) burst.data[i] = data[i];
bursts[0] = burst;// this pushes the burst directly downstream because
// it is a sequence of bursts
shortBurstPort->pushBursts(bursts);

2、突发信号相关信息 (SRI)

BurstSRI 对象随每个数据突发一起传送，并描述数据生产者的数据负载和处理状态。下表仅描述了在资源之间传递突发数据时数据结构的必需字段。

3、多输出端口

每个输出突发输入/输出（BurstIO）端口类型都提供了基于流 ID 和连接 ID 过滤来自资源的突发数据的能力。要使用端口的多出能力，资源必须包含类似于以下的代码：

<structsequence id="connectionTable"><struct id="connectionTable::connection_descriptor"name="connection_descriptor"><simple id="connectionTable::connection_id" name="connection_id"type="string"><kind kindtype="configure"/></simple><simple id="connectionTable::stream_id" name="stream_id" type="string"><kind kindtype="configure"/></simple><simple id="connectionTable::port_name" name="port_name" type="string"><kind kindtype="configure"/></simple></struct><configurationkind kindtype="configure"/>
</structsequence>

为了将特定的数据流引导到特定的连接，需要将 connectionTable 对象传递给端口的 updateConnectionFilter 方法。当路由模式设置为 ROUTE_CONNECTION_STREAMS 时，端口将会将过滤状态应用于通过资源的 BurstIO 端口传出的任何突发流量。要将突发传递给现有连接，端口的过滤器表中必须存在与下游资源的突发的流 ID 和连接 ID 匹配的项。

4、使用复数数据

每个传入数据的 BurstPacket 提供了 getComplex()` 方法，以表示向量是否包含复数样本（它由实部和虚部组成）。复数数据以交替的实部和虚部值发送。开发者可以以任何方式处理这些数据；然而，本节描述了将数据转换为更易于处理形式的常用方法。

①、在 C++ 中转换复数数据

在 C++ 中，传入的突发输入/输出（BurstIO）数据向量可以被强制转换为复数值的 std::vector。例如：

BurstShortIn::BurstPacket *pkt = myShortPort->getPacket(bulkio::Const::BLOCKING);
if ( pkt->isComplex() ) {BurstShortIn::ComplexType cplx_data = pkt->getComplexData();// ... do some processing with cplx_data
}

5、时间戳

以下代码段提供了一个示例，展示如何构造一个要在突发信号相关信息（SRI）中发送的 BULKIO::PrecisionUTCTime 时间戳。

/*** To create a time stamp from the current time of day*/BULKIO::PrecisionUTCTime tstamp = burstio::utils::now();

6、端口统计

所有突发输入/输出（BurstIO）端口支持批量输入/输出（BulkIO）统计接口，并添加了跟踪特定于突发的指标的额外关键词。统计数据是在 10 次 pushBurst 调用的窗口中跟踪的。输入端口包含单个 PortStatistics 结构，而输出端口包含一系列 PortStatistics 结构；每个连接一个结构。有关 BULKIO::PortStatistics 的更多信息，请参见端口统计。输入和输出端口的额外 BurstIO 指标在以下表格中描述：

①、C++

以下示例说明了一个组件，该组件对传入的突发数据进行转换，并将结果向下游推送。

burstio::BurstShortIn::PacketType *pkt;
pkt = inShortPort->getPacket(bulkio::Const::NON_BLOCKING);// check if a valid packet was returned
if ( pkt == NULL ) {return NOOP;
}// check for EOS
if ( pkt->getEOS() ) {outShortPort->pushBurst(pkt->getSequence(), pkt->getSRI(), pkt->getEOS());
}// do some processing.....to the burst contents
BurstShortOut::SequenceType  data =  do_some_magic(pkt->getSequence());// we changed the data so calc new time stamp....
BULKIO::PrecisionUTCTime newTS = calc_timestamp(pkt->getTime());  outShortPort->pushBurst(data, pkt->getSRI(), newTS, pkt->getEOS());

二、消息传递

消息传递依赖于 CORBA 的事件结构作为传输结构。在 CORBA 的事件 API 中，消息通过使用函数 push() 以 Any 类型传递。

虽然 CORBA 管理数据的编组和传递，但它并未提供任何固有于事件的机制来描述 Any 类型的内容。REDHAWK 决定利用现有的负载结构描述符来描述消息的负载，即属性接口描述语言（IDL）。选择此接口消除了创建描述消息的新 IDL 的需要。此外，已有一种 XML 结构映射到高效的二进制数据结构，允许使用 XML 来描述消息内容，同时消除了在消息传递机制中引入 XML 解析器的需要。

为了支持这项额外功能，REDHAWK 扩展了属性描述符，允许属性具有消息类型。唯一可以具有有效消息类型的属性是结构。

1、消息生产者

在创建新组件或编辑现有组件时，可以创建一个消息生产者。创建消息生产者后，您必须注册您的代码，以便从端口发送消息。以下程序解释了如何创建消息生产者并发送消息。

①、创建一个消息生产者

使用 REDHAWK IDE 向组件或设备添加消息生产者端口，请遵循以下步骤：

• 1）从项目资源管理器视图中，双击组件的软件包描述符（SPD）文件。此时将显示组件编辑器。

• 2）在组件编辑器中，选择“属性”标签。组件编辑器的属性标签页将被显示。
  

• 3）要添加结构属性，请点击“添加结构”。此时将显示属性标签的结构属性部分。
  

• 4）在结构属性部分，输入产生的消息的名称。ID 默认为您输入的名称。从“类型（Kind）”下拉菜单中，选择“消息（message）。
  

• 5）在“所有属性（All Properties）”部分，选择您添加的结构属性。默认情况下，已经存在一个简单成员。您可以修改它并为结构属性创建额外的成员。有关属性结构、类型和种类的更多信息，请参考管理和定义属性。修改和/或创建这些成员后，结构的成员和相应的属性信息将被显示。
  

• 6）选择“端口（Ports）”标签，点击“添加（Add）”，并在“名称（Name）”字段中输入名称。在“端口详情（Port Details）”部分，在“方向（Direction）”下拉菜单中选择“出（out ）”。在“接口（Interface）”字段旁，点击“浏览（Browse）”。将显示“选择接口（Select an Interface）”对话框。从显示的接口列表中，选择“ExtendedEvent> MessageEvent”并点击“确定（OK）”。此时将显示消息生产者端口的信息。
  

• 7）重新生成组件。创建消息生产者后，您可以从消息生产者端口发送消息。

②、发送消息

以下代码示例演示了如何在 C++中从组件的消息输出端口向事件通道或另一个组件的消息输入端口发送外发消息。

出于以下示例的目的，假设结构如下：

• id: foo
• 包含两个成员：
  • name: some_string, type: string
  • name: some_float, type: float
• 组件的 uses 端口称为 message_out
• 组件的名称是 message_producer
  
  
  
  

在每个示例中，通过声明该类型的变量来创建消息。然后，设置其状态，并使用消息端口的 sendMessage() 方法发送消息，消息变量作为参数。

C++：
要生成消息，可以在实现文件的 serviceFunction() 方法中添加以下代码。

foo_struct my_msg;
my_msg.some_string = "hello";
my_msg.some_float = 1.0;
this->message_out->sendMessage(my_msg);
// Send a message to a specific connection by providing a `connectionId` parameter.
// If `connectionId` does not match any connection, an `std::illegal_argument` exception is thrown.
this->message_out->sendMessage(my_msg, "connection_1");

2、消息消费者

在创建新组件或编辑现有组件时，可以创建消息消费者。创建消息消费者后，您必须注册您的代码以从端口接收消息。以下程序解释了如何创建消息消费者并注册代码以处理消息。

①、创建消息消费者

流程同上消息生产者流程，与创建消息消费者流程不同之处只是在端口详细信息部分的方向下拉列表中，选择bi-dir<uses/provides>

②、注册接收消息

以下示例解释了如何在 C++中注册代码以处理传入消息。

出于以下示例的目的，假设结构如下：

• id: foo
• 包含两个成员：
  • name: some_string, type: string
  • name: some_float, type: float
• 组件的 uses/provides 端口称为 message_in
• 组件的此消息的回调函数为 messageReceived()
• 组件的名称是 message_consumer

如果此组件与消息生产者或事件通道之间存在连接，则以下代码示例将处理传入消息。

任何带有属性 ID foo 的消息都将触发回调函数 messageReceived()。

C++ ：
鉴于事件的异步性质，为消费者选择了回调模式。在组件头文件中，声明以下回调函数：

void messageReceived(const std::string &id, const foo_struct &msg);

在组件源文件中，实现回调函数：

void message_consumer_i::messageReceived(const std::string &id, const foo_struct &msg) {LOG_INFO(message_consumer_i, id<<" "<<msg.some_float<<" "<<msg.some_string);
}

在构造器（constructor()）方法中，注册回调函数：

message_in->registerMessage("foo", this, &message_consumer_i::messageReceived);

3、查看消息

消息是事件，其有效负载定义与组件属性中的结构相关联。查看消息可以使用用于查看事件的相同技术。
要在终端窗口查看发送到事件通道的事件和消息：

eventviewer <domain name> <event channel>

实用程序的帮助:

eventviewer --help

eventviewer REDHAWK_DEV testchan
示例输出：

这里出现了错误，还未找到解决办法，待后续找到后修改

4、连接生产者和消费者

生产者和消费者可以点对点连接，也可以通过IDE 中的事件通道连接。将生产者直接连接到消费者不需要应用程序，并且可以在沙箱中完成：

from ossie.utils import sb
sb.catalog()
#['structs_test', 'm_in', 'prop_changes', 'm_out','pass']
prod=sb.launch("m_out")
cons=sb.launch("m_in")
prod.connect(cons)
#True
sb.start()

这里出现了错误，还未找到解决办法，待后续找到后修改

5、连接回调

在组件中，许多 REDHAWK 中的 Uses 端口类支持在连接建立或断开时进行通知。支持的端口类型和语法因语言而异。

①、C++

在 C++ 中，Bulk 输入/输出（BulkIO）、Burst 输入/输出（BurstIO）和消息 Uses 端口的连接通知机制已标准化。连接和断开回调被注册在端口上。

以下示例假设一个 C++ 组件具有 BulkIO 浮点输出端口，dataFloat_out；然而，对于 BurstIO 和消息端口，语法是相同的。

在组件头文件中，将回调声明为私有成员函数。连接和断开回调都接收单一参数，连接 ID（一个引用的 std::string）：

void dataFloatConnected(const std::string& connectionId);
void dataFloatDisconnected(const std::string& connectionId);

在组件源文件中，实现回调函数：

void MyComponent_i::dataFloatConnected(const std::string& connectionId)
{LOG_INFO(MyComponent_i, "New connection " << connectionId << " on dataFloat_out");
}void MyComponent_i::dataFloatDisconnected(const std::string& connectionId)
{LOG_INFO(MyComponent_i, "Disconnected " << connectionId << " on dataFloat_out");
}

然后，在组件 constructor() 中注册回调函数

dataFloat_out->addConnectListener(this, &MyComponent_i::dataFloatConnected);
dataFloat_out->addDisconnectListener(this, &MyComponent_i::dataFloatDisconnected);

addConnect Listener()和 addDisconnectListener() 接受两个参数：目标对象（通常是this）和一个指向成员函数的指针。

没有必要同时注册连接和断开回调。

6、自定义 IDL 接口

REDHAWK提供前端接口（FEI）和标准核心框架（CF）接口（如 CF::Resource），用于控制实体并促进互操作性。在某些用例中，您可能会发现需要使用自定义接口描述语言（IDL）来控制实体。对于这些用例，您可以在 IDE 中创建自定义 IDL 项目。

将端口从 FEI 接口或自定义 IDL 接口添加到组件或设备中，允许该实体通过 CORBA 控制其他实体。由于这些端口的通用性质，不可能创建像 BulkIO 这样的语言映射，因此交互是通过标准的 CORBA API 进行的，其完整描述超出了本手册的范围。然而，REDHAWK 代码生成器将生成简化与端口交互的端口。以下部分解释了使用（输出）端口的情况，因为它们最有可能被生成，例如，用于控制 FEI 设备。

①、连接反馈

在所有三种支持的语言中，FEI、标准 CF 或自定义 IDL 端口将有所有方法和属性映射到端口，并且端口将委托调用给远程连接。在 REDHAWK 中，端口可能没有连接、有一个连接或多个连接。这些条件中的每一个都可能为使用端口进行通信的人创建问题；例如，如果发送了一个控制请求并且没有连接，那么用户应该被告知请求没有发送到任何地方。

同时，并非所有方法都是相同的。一些方法只在一个方向上推送数据，一些方法有返回值，一些方法有作为指针的参数，需要用信息填充（输出或输入/输出参数）。当调用端口方法并且端口无法进行调用或调用无法明确时（例如，如果存在两个连接并且函数包含返回值），则在用户代码中引发 PortCallError。下表描述了满足的方法签名标准及其对应的行为。

如果一个方法作为其非异常 API 的一部分有任何形式的返回值（表现为非 void 返回值，或一个输出或输入/输出参数），那么如果端口有多于一个的连接，就会引发一个异常。此外，如果尝试调用时实际上没有连接，就会引发一个错误。

②、连接选择

虽然当所需的连接不明确时，生成的端口类会触发一个错误，但它也包含一个 API，允许开发者选择应该使用哪个连接。每个方法都有一个可选参数，connection_id，允许调用者明确应该使用哪个连接。默认值行为将使用最后建立的连接。如果指定的 connection_id 不存在，将引发 PortCallError。

在以下部分中，用于明确连接的相同模式在所有三种支持的语言中都提供了。

以下代码示例使用调用 CF::File 接口的 read 方法时的默认行为。

 CF::OctetSequence_var _data = new CF::OctetSequence();CF::OctetSequence_out data(_data);this->file_out->read(data, 10); // read 10 characters from the last connection made to the port

以下代码示例将读取调用明确指向一个特定的连接，connection1。

 CF::OctetSequence_var _data = new CF::OctetSequence();CF::OctetSequence_out data(_data);this->file_out->read(data, 10, "connection1"); // read 10 characters from the connection called "connection1"

要查看可用的连接，请使用以下代码：

 std::vector<std::string> _connection_ids = this->file_out->getConnectionIds();

③、方法映射

方法名称映射遵循在连接选择中描述的模式；即方法名称与 IDL 中描述的名称相同，但增加了一个额外的参数（可选地使用），可以指定应使用哪个连接。属性被映射为对 CORBA 对象的函数。REDHAWK 提供额外的 API 来消除对多个连接的调用的歧义。

④、读取属性

通过调用属性的名称作为函数来执行读取属性。例如，如果端口 my_port 包含字符串属性 greeting，则可以如下检索 greeting 的值：

 std::string _greeting = this->my_port->greeting();

要从特定连接检索值，需要_get_前缀：

 std::string _greeting = this->my_port->_get_greeting("some_connection_name");

⑤、写入属性

在 C++ 中写入属性涉及使用适当的参数调用函数：

 this->my_port->greeting("hello"); // write "hello" to the attribute "greeting"this->my_port->greeting("hello", "some_connection_name"); // write "hello" to the attribute "greeting" over connection "some_connection_name"

---

我的qq：2442391036，欢迎交流！

---