Message Publishing

Principles

Figure 1 shows the process of invoking key interfaces for publishing a common message.

Figure 1 Process of invoking key interfaces for publishing a common message

Initialize resources.
Before invoking the OpenHiva interface, invoke the OpenHiva::Init interface for initialization. If the return value is 0, the initialization is successful. Otherwise, the initialization fails.
Initialization includes node name registration, thread group creation, and resource application.
- Node name registration:
  - The names of different nodes must be unique. Otherwise, the nodes that are started later will fail to be initialized.
  - If the initialization fails, the node cannot perform the publishing or subscription action.
- Thread group creation:
  When the OpenHiva::Init interface is invoked, the thread group parameter needs to be transferred. Because the publisher does not need to set the callback function, an empty thread group can be transferred for initialization.
Register the publishing information.
1. Before registering the publishing information, you must create the node handle OpenHiva::Node n.
2. Use the created node handle n to invoke the CreatePublisher interface to register the topic to be published with the DataMaster. After the CreatePublisher interface is invoked, the Publisher object is returned.
3. Invoke the Ready interface of the Publisher object to determine whether the publisher is successfully created.
When the CreatePublisher interface is invoked, OpenHiva creates a memory pool and publish queue, registers information such as the topic name and publish queue ID with the central node (DataMaster), and binds the publish queue ID to the ID of the subscription queue that subscribes to the topic.
Publish the message.
The user uses the Publisher object returned in Step 2 to invoke its Publish interface, fill the data in the publish queue, and publish the message. If the topic has a subscriber, the queue scheduler transfers the message from the publish queue to the bound subscription queue based on the queue binding relationship in step 2. Then the event scheduler wakes up the working thread of the subscriber to process the message in the subscription queue.
- If the queue binding relationship is not created, OpenHiva determines the processing based on the input parameter topicOptions.overWriteFlag of the CreatePublisher interface after the publish queue or the corresponding data block is full.
  - If overWriteFlag is set to False (default value), the file is not overwritten. In this case, the system discards the new data to be enqueued and retains the old data.
  - If overWriteFlag is set to True, data is overwritten. In this case, the system discards the old data in the queue and saves the new data.
- The OpenHiva internal message dequeue processing varies depending on the value of the input parameter topicOptions.queueFCFlag of the CreatePublisher interface.
  - If queueFCFlag is set to False (default value), queue flow control is disabled. In this case, the input parameter queueTTL of the interface is invalid, and messages in the queue are not discarded when messages are dequeued.
  - If queueFCFlag is set to True, queue flow control is enabled. In this case, the input parameter queueTTL of the interface is valid (the default value is 1000 ms). When message are dequeued, the message whose storage duration exceeds the value of queueTTL is discarded. If the storage duration of all messages in the queue exceeds the value of queueTTL, the latest message is retained and other messages are discarded.
Release resources.
The corresponding resources need to be released before the process ends. The defined OpenHiva APIs cannot be used.
- In the exception branch, the OpenHiva::Shutdown interface needs to be invoked to release resources, including deregistering the queue ID and topic information and releasing the corresponding memory block.
- In the main thread, the OpenHiva::WaitForShutdown interface needs to be proactively invoked to release resources to prevent the main thread from exiting in advance. When the process stops abnormally, the WaitForShutdown interface automatically invokes the Shutdown interface to clear resources.

Sample Code

Common and relatively stable parameters (such as queueSize) in the code can be managed in a unified manner through the configuration management module to facilitate subsequent maintenance. For details, see Configuration Management.

The following is a code example of key steps in publishing a common message, which is for reference only:

#include "open/publisher.h"
#include "open/init.h"
#include "open/node.h"
#include "std_msgs/include/StringMessage.h"
#include "Hiva_time/Hiva_rate.h"

// FixMessage is a user-defined function used to process published messages.
void FixMessage(Hiva::StdMsgs::StringMessage &msg)
{
    std::stringstream ss;
    ss << "talker: ==>Hello World " << count;
    msg.stringData = ss.str();          // Assign the string content to the message. The specific value depends on the user message structure.
    HIVA_WARN("%s", msg.data.c_str());
}

int main(int argc, char **argv)
{
    // 1. Initialize resources.
    uint32_t queueSize = 10;          // Length of the data publish queue. The maximum value is 128.
    uint32_t maxMsgSize = 20000;     // Maximum size of a frame, in bytes.
    uint32_t blockNum = 20;         // Number of blocks in the shared memory pool. The default value is queueSize x 2. The maximum value is 256.
    uint32_t queueSize = 10U;      // Queue length.
    bool overwrite = true;         // Indicates whether to overwrite data.
    bool queueFCFlag = false;     // Indicates whether to enable queue flow control.
    uint16_t queueTTL = 0U;       // Time difference of flow control
    std::string topicName = "/chatter";  // Topic published by the publisher.
    std::string nodeName = "/talker";     // Publisher node name.
    // Define a thread group.
    std::vector<OpenHiva::ScheduleGroup> threadGroup;
    
    
    
    
    // Invoke the resource initialization interface.
   OpenHiva::Init(argc, argv, threadGroup);  
    HIVA_EVENT("Talker init ok!");

    // 2. Register and publish messages.
    // Construct the Node object.
    OpenHiva::Node n(nodeName);
    // Construct TopicOptions.
    OpenHiva::TopicOptions topicOps;
    topicOps.BuildGroupName(testGroup.groupName)
            .BuildMessageTraits<Hiva::StdMsgs::StringMessage>()
            .BuildShmOptions(maxMsgSize, blockNum)
            .BuildQueueOptions(queueSize, overwrite, queueFCFlag, queueTTL)
            .BuildTopicName(topicName);
    // Invoke the CreatePublisher interface through the NodeHandle object to publish the topic. The Publisher object is returned.
    std::shared_ptr<OpenHiva::Publisher> chatterPub = n.CreatePublisher<Hiva::StdMsgs::StringMessage>(topicName, topicOps);  
    // Check whether the Publisher object is successfully constructed. If the operation fails, invoke the Shutdown function to release resources.
    if ((chatterPub == nullptr) || !chatterPub->Ready()) {
        HIVA_ERROR("chatterPub Ok() Fail!");
        OpenHiva::Shutdown(); 
        return 0;
    }
    HIVA_INFO("Publisher creat ok");

    // 3. Publish a message.
    int count = 0;    
    // Check the Hiva node status. If the node is enabled, true is returned. If the node is shut down or fails to be initialized, false is returned. In this case, packets cannot be sent or received correctly.
    while (OpenHiva::Ready()) {  
        Hiva::StdMsgs::StringMessage msg;  // Define the message to be published.
        FixMessage(msg);                  // Preprocess the message to be published.
        chatterPub.Publish(msg);        // Publish a message to the publish queue.
        sleep(1);
        ++count;
    }

    // 4. Release resources.
    // This function is invoked in blocking mode to prevent the process from exiting. When you press Ctrl+C or use the kill-2 command on the process, the function returns immediately.
    OpenHiva::WaitForShutdown();  
    return 0;
}

Parent topic: Common Messages