Overview

Subscription-Publishing Principle

The OpenHiva supports topic-based communication and the following message types:

Common message: provides standard topics for message subscription and publishing.
The message memory is applied for and released by OpenHiva. You only need to pay attention to the service logic. The code logic is relatively simple.

This chapter uses message publishing and subscription as an example to describe how to use Common Messages.
BufferMessage: provides the HivaBuffer for message subscription and publishing.
The message memory is applied for and released by the user, and the data in the memory is managed by the user. The message must be in the fixed BufferMessage format. The framework obtains the memory address applied by the user from the published message. If an error occurs during the publishing, the framework automatically releases the memory. If the message is published successfully, the memory is automatically released by the subscription end. Users do not need to pay attention to the memory release.

BufferMessage applies to image message communication scenarios. Especially for data obtained from sensors such as lidar and camera, the data volume is large. To reduce the copy of image data, BufferMessage can be used to improve transmission performance.

This chapter uses the basic scenarios of message publishing and subscription as an example to describe how to use BufferMessage (Basic Scenario). Then, this chapter uses the inference scenario and NN engine as an example to describe how to use the BufferMessage (Asynchronous Inference Scenario) to obtain data, perform inference, and publish inference results.

The NN engine is implemented inside a node. After the inference model is loaded to an application and the original camera data is input, the inference result can be published in a topic or BufferMessage format.

The principles of message publishing and subscription are basically the same regardless of the message type. Figure 1 shows the key message flows. The Talker publishes topics as a publisher, and the Listener subscribes to topics as a subscriber. The Talker and Listener communicate with each other through the DataMaster (central node). The QueueScheduler functions as the message queue scheduler. The pipe name is the topic name. The pipe supports single-input multiple-output.

Figure 1 Message subscription-publishing process

The publishing end registers the topic to be published and the TopicOptions attributes (such as the queue depth and maximum message size) corresponding to the topic with the DataMaster through the CreatePublisher interface.
The subscription end registers the topic to be subscribed to, callback function, and TopicOptions attributes (such as the queue depth, maximum message size, and subscriber group) corresponding to the topic with the DataMaster through the CreateSubscriber interface.
The OpenHiva binds the publishing queue ID and subscription queue ID based on the registered queue information received by the DataMatser and stores them in the QueueScheduler.
The publishing end publishes messages of a specified topic through the Publish interface.
The subscription end uses the QueueScheduler to transfer messages from the publishing queue to the bound subscription queue and wakes up the working thread of the subscription node.
The subscription thread invokes the callback function to process the message and obtain the published topic message.

TopicOptions Parameter Description

The TopicOptions parameter is a set of attribute values of a topic message, including the topic name, topic communication mode, queue depth, and message size. Currently, the following TopicOptions parameter configuration methods are supported:

TopicOptions configuration priority: YAML file > parameter input > default value
If both the YAML file and parameter input are configured, the parameters in the YAML file are used and the interface input parameters become invalid.

Default value: If a parameter is not configured, OpenHiva provides a feasible default parameter value.
Parameter input: Parameters are configured through a series of Build interfaces.

YAML file: Parameters are configured through the YAML file. After the configuration is complete, use the __transport=yaml path parameter in the command line to specify the YAML file path for the configuration file to take effect.

The following is a YAML file configuration example. For details about the parameters, see Table 1.

testTopic:                 
    transport: [Shm, Dsf] 
    groupName: test_group   
    blockSize: 102400      
    blockNum: 10           
    Dsf:
        queueDepth: 5      
        overwrite: false  
        queueFCFlag: false 
        queueTTL: 0      
    Dds:
         history: DDS_QOS_HISTORY_POLICY_KEEP_LAST
         depth: 5
         reliability: DDS_QOS_RELIABILITY_POLICY_BEST_EFFORT
         durability: DDS_QOS_DURABILITY_POLICY_VOLATILE
         deadline:
             second: 100
             nanosecond: 
         lifespan:
             second: 10
             nanosecond: 10
         liveliness: DDS_QOS_LIVELINESS_POLICY_AUTOMATIC
         lease_duration:
             second: 
             nanosecond:

**Table 1** Parameters in the YAML configuration file
Category	Parameter	Description
Basic parameter	testTopic	Specifies the name of the topic for which parameters need to be configured.
	transport	Configures the communication mode corresponding to the topic. You can specify one or more communication modes, which must be enclosed in square brackets ([]). NOTE: Currently, the following communication modes are supported: Dsf, Vfs, Shm, and Dds. If the configured communication mode is invalid, Shm is used by default in the service scenario, Vfs is used by default in the management plane, and Dsf is used by default in the data plane.
	groupName	Specifies the name of the group to which the topic belongs.
	blockSize	Specifies the size of a memory block. Memory resources are applied for based on blockSize during Dsf or Shm communication.
	blockNum	Specifies the number of memory blocks. When the Dsf or Shm communication is used, multiple memory blocks are applied for during initialization. The size of each memory block is the value of blockSize, but the value must meet the following requirement: blockSize x blockNum < 300 MB.
Dsf	queueDepth	Specifies the length of the queue for caching messages during communication.
	overwrite	Specifies whether to overwrite data. The value can only be true or false. If this parameter is set to true, messages at the beginning of the receive queue is discarded and the newly received message is added to the end of the receive queue when the receive queue is full. If this parameter is set to false, the newly received message is discarded when the receive queue is full.
	queueTTL	Specifies whether to enable queue flow control. The value can only be true or false. If this parameter is set to true, queue flow control is enabled and the queueTTL parameter takes effect. If this parameter is set to false, queue flow control is disabled and the queueTTL parameter does not take effect.
	queueTTL	Specifies the queue flow control duration, in milliseconds. This parameter is valid only when Dsf/queueFCFlag is set to true. When queueTTL is 0, a maximum of one frame can be buffered in the receive queue. When queueTTL is greater than 0, the time difference between the time when a message enters the receive queue and the time when the message leaves the receive queue is calculated. If the time difference exceeds the value of queueTTL, the message is discarded.
Dds NOTE: DDS QoS policies can be separately configured on the publishing end and receiving end. The two ends can match only when the policies are compatible. DDS communication node discovery is affected by network fluctuation. When a single device is unloaded, it takes about 500 ms at most to match the underlayer communication node. In the default QoS configuration, packets sent before node matching are discarded. For details about the compatibility of different policy combinations, see Table 2.	history	Specifies DDS message storage policy. The options are as follows: DDS_QOS_HISTORY_POLICY_KEEP_LAST: reserves the most recent 'depth' of samples. DDS_QOS_HISTORY_POLICY_KEEP_ALL: reserves all message samples.
	depth	Specifies the queue length. This parameter is valid only when history is set to DDS_QOS_HISTORY_POLICY_KEEP_LAST.
	reliability	Specifies the message reliability. The options are as follows: DDS_QOS_RELIABILITY_POLICY_RELIABLE ensures that the message is delivered. DDS_QOS_RELIABILITY_POLICY_BEST_EFFORT: Messages are not retransmitted and no arrival acknowledgment of the transmission is required.
	durability	Specifies the message durability. The options are as follows: DDS_QOS_DURABILITY_POLICY_VOLATILE: Messages sent are ignored. Messages are received only when the datareader matches the datawriter. DDS_QOS_DURABILITY_POLICY_TRANSIENT_LOCAL: After a new datareader is created, the messages that have been sent are added to the historical records.
	deadline	Specifies validity period for sending a message to the receiver. After a message is sent, the receiver checks the validity period. If the validity period expires, the receiver considers the message invalid and discards it. Second Nanosecond The total time is the sum of the two times.
	lifespan	Specifies the validity period of a message in the system. Second Nanosecond The total time is the sum of the two times.
	liveliness	Specifies the life cycle detection policy of the datareader and datawriter. The options are as follows: DDS_QOS_LIVELINESS_POLICY_AUTOMATIC: The system automatically verifies the validity of the datareader and datawriter. DDS_QOS_LIVELINESS_POLICY_MANUAL_BY_TOPIC: The publishing end checks the validity of the subscription end within the lease duration. If the subscription end does not respond to the heartbeat signal sent by the publishing end, the subscription end is considered invalid.
	lease_duration	Specifies the time when the validity of the subscription end and publishing end is checked. If the subscription or publishing end responds to the heartbeat signal after the time specified by this parameter, the subscription or publishing end is considered invalid. Second Nanosecond The total time is the sum of the two times. This parameter must be used together with DDS_QOS_LIVELINESS_POLICY_MANUAL_BY_TOPIC.

**Table 2** DDS QoS policy compatibility
Policy	Publishing End	Subscription End	Compatible or Not
Reliability policy compatibility	Best effort	Best effort	Compatible
	Best effort	Reliable	Incompatible
	Reliable	Best effort	Compatible
	Reliable	Reliable	Compatible
Durability policy compatibility	Volatile	Volatile	Compatible
	Volatile	Transient local	Incompatible
	Transient local	Volatile	Compatible
	Transient local	Transient local	Compatible
Deadline policy compatibility	Default value (QOS_DURATION_MAX)	Default value (QOS_DURATION_MAX)	Compatible
	Default value (QOS_DURATION_MAX)	x	Incompatible
	x	Default value (QOS_DURATION_MAX)	Compatible
	x	x	Compatible
	x	y (y > x)	Compatible
	x	y (y < x)	Incompatible
Lifespan policy compatibility	N/A	N/A	N/A
Liveliness policy compatibility	Automatic	Automatic	Compatible
	Automatic	Manual by topic	Incompatible
	Manual by topic	Automatic	Compatible
	Manual by topic	Manual by topic	Compatible
Lease duration policy compatibility	Default value (QOS_DURATION_MAX)	Default value (QOS_DURATION_MAX)	Compatible
	Default value (QOS_DURATION_MAX)	x	Incompatible
	x	Default value (QOS_DURATION_MAX)	Compatible
	x	x	Compatible
	x	y (y > x)	Compatible
	x	y (y < x)	Incompatible
Note: Assume that x and y are valid values. The value range is [0, QOS_DURATION_MAX].

Parent topic: Message Subscribing-Publishing