Message Publishing

Principles

Figure 1 shows the process of invoking key interfaces for publishing BufferMessage messages.

Figure 1 Process of invoking the BufferMessage message publishing interface

1. Initialize resources.

   The process is similar to the process of Common Messages > Message Publishing > Initialize resources. You must invoke the OpenHiva::Init interface to initialize resources.

2. Register the messages to be published.

   The amount of data obtained from sensors such as cameras is large and unstructured. Therefore, the transmission format must be the fixed message format BufferMessage.

   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 does not need to create a memory pool. Instead, the OpenHiva directly uses the shared memory applied by the corresponding sensor driver for transfer. Therefore, maxMsgSize can be set to 0.

3. Publish the message.

   Before using the Publisher object returned in step 2 to invoke the Publish interface to publish messages, you need to apply for a HivaBuffer and fill data in the HivaBuffer as required. The key process is as follows:

   1. Initialize the memory pool by invoking the HivaBufferPool::InitMemoryPool interface, and enter the memory pool name, size of each memory block, and number of memory blocks.
   2. Invoke the HivaBufferPool::Allocate interface to apply for a HivaBuffer from the memory pool. The size of the HivaBuffer memory is the same as that of the memory block in the memory pool.
   3. Invoke the HivaBuffer::GetBuff interface to obtain the memory address dataPtr and dataLen.
   4. Write data to the address dataPtr. Note that the length cannot exceed dataLen.
   5. After data is written, invoke HivaBuffer::SetBuffDataLen to set the length of the written data.
   6. (Optional) If you want to transmit some private data with messages, you can invoke the HivaBuffer::SetUserData interface.
   7. Invoke the OpenHiva::BufferMessage interface to construct the BufferMessage object.

   After the preceding steps are complete, you can invoke the Publish interface to publish BufferMessage messages.

4. Release resources.

   The process is similar to the process of Common Messages > Message Publishing > Release 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 the key steps for publishing a BufferMessage message in a basic scenario, which is for reference only:

#include <string>
#include <iostream>
#include <memory>
#include <sstream>
#include "open/publisher.h"
#include "open/init.h"
#include "open/node.h"
#include "open/Hiva_buffer_pool.h"
#include "open/buffer_message.h"
#include "Hiva_time/Hiva_rate.h"

void Usage()
{
    std::cout << "Usage: pmupload test_create_buffer_pub nodeName topicName groupName bindType maxMsgSize queueSize blockNum overwrite queueFCFlag queueTTL rate msgCnt transport" << std::endl;
    std::cout << "if argc==2, pub will use default value: pmupload test_create_buffer_pub testNodeBufferPub testTopic testGroupBufferPub 0 102400 10 10 1 0 0 1 10" << std::endl;
    std::cout << "bindType can be 0 1 2" << std::endl;
    std::cout << "Pointing: req/res size must bigger than messageSize + 8, or serialize will fail" << std::endl;
}

int main(int argc, char **argv)
{
    // 1. Initialize resources.
    std::string nodeName = "testNodePub";
    std::string topicName = "testTopic";
    std::string groupName = "testGroupPub";
    OpenHiva::ScheduleType scheType = OpenHiva::ScheduleType(0);
    uint32_t maxMsgSize = 100 * 1024U;
    uint32_t queueSize = 10U;
    uint32_t blockNum = 10U;
    bool overwrite = true;
    bool queueFCFlag = false;
    uint16_t queueTTL = 0U;
    uint32_t rate = 1U;
    uint32_t msgCnt = 10U;
    std::string transport;
    int argNum = 14;
    if (argc < argNum) {
        Usage();
        if (argc != 2) {
            return 0;
        }
    } else {
        nodeName = argv[1];
        topicName = argv[2];
        groupName = argv[3];
        scheType = (OpenHiva::ScheduleType)strtol(argv[4], NULL, 10);
        maxMsgSize = (uint32_t)strtol(argv[5], NULL, 10);
        queueSize = (uint32_t)strtol(argv[6], NULL, 10);
        blockNum = (uint32_t)strtol(argv[7], NULL, 10);
        overwrite = (bool)(strtol(argv[8], NULL, 10) & 0x01);
        queueFCFlag = (bool)(strtol(argv[9], NULL, 10) & 0x01);
        queueTTL = (uint16_t)strtol(argv[10], NULL, 10);
        rate = (uint32_t)strtol(argv[11], NULL, 10);
        msgCnt = (uint32_t)strtol(argv[12], NULL, 10);
        transport = argv[13];
    }
    // Define a thread group.
    std::vector<OpenHiva::ScheduleGroup> scheGrpVec;
   
    
   
   
    // Invoke the resource initialization interface.
    OpenHiva::Init(argc, argv, scheGrpVec);
    HIVA_EVENT("publisher init ok!");

    // 2. Register and publish messages.
    // Construct the Node object.
    OpenHiva::Node node(nodeName);
    // Construct TopicOptions.
    OpenHiva::TopicOptions topicOps;
    topicOps.BuildGroupName(groupName)
            .BuildMessageTraits<OpenHiva::BufferMessage>()
            .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> pub = node.CreatePublisher<OpenHiva::BufferMessage>(topicName, topicOps); 
    // Check whether the Publisher object is successfully constructed. If the operation fails, invoke the Shutdown function to release resources and exit.
    if ((pub == nullptr) || (!pub->Ready())) {
        HIVA_ERROR("create publisher failed");
        OpenHiva::Shutdown();
        return 0;
    }
    HIVA_EVENT("create publisher success");

    // 3. Publish a message.
    uint32_t count = 0U;
    Hiva::HivaRate interval(rate);
   // Initialize the memory pool.
    OpenHiva::HivaBufferPool HivaBufferPool;
    uint32_t ret = HivaBufferPool.InitMemoryPool(topicName, maxMsgSize, queueSize*2);       
    if (ret != Hiva::HIVA_SUCCESS) {
        HIVA_ERROR("create buff pool failed");
    }
    else {
        HIVA_INFO("create pool success");
    }
    // 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()) {
        std::stringstream ss;
        ss << "talker: ==>Hello World " << count;
        OpenHiva::HivaBuffer Hivabuffer;
        // Apply for a HivaBuffer.
        ret = HivaBufferPool.Allocate(Hivabuffer);
        if (ret != Hiva::HIVA_SUCCESS){
            HIVA_ERROR("alloc new buff failed, return code is %u", ret);
            continue;
        } else {
            HIVA_INFO("create mbuf success");
        }
        void *currBlockPtr = nullptr;
        size_t dataSize = 0U;
        // Obtain the memory address and length of the written data.
        ret = Hivabuffer.GetBuff(currBlockPtr, dataSize);
        if (ret != Hiva::HIVA_SUCCESS) {
            HIVA_ERROR("GetBuff failed and return %u", ret);
            continue;
        }
        *reinterpret_cast<uint32_t *>(currBlockPtr) = ss.str().size();
        ret = memcpy_s(reinterpret_cast<uint8_t *>(currBlockPtr) + 4, dataSize - 4,
            ss.str().c_str(), ss.str().size());
        if (ret != Hiva::HIVA_SUCCESS) {
            HIVA_ERROR("memcpy failed and return %u", ret);
            continue;
        }
        HIVA_WARN("%s", ss.str().c_str());
        // Construct the BufferMessage object and publish the BufferMessage message.
        pub->Publish(OpenHiva::BufferMessage(Hivabuffer));
        interval.Sleep();
        ++count;
        if (count > msgCnt) {
            break;
        }
    }

    // 4. Release resources.
    OpenHiva::WaitForShutdown();
    return 0;
}