Message Subscription

Principles

Figure 1 shows the process of invoking key interfaces for subscribing to common messages.

Figure 1 Process of invoking the common message subscription interface

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 common message subscription, which is for reference only. When a user creates a callback thread, the internal processing of OpenHiva varies according to the value of ThreadGroup.scheduleType in the thread group.

• In non-USER_DEFINED mode, the subscriber can automatically start a thread to obtain subscription messages.
• In USER_DEFINED mode, the subscriber does not automatically start a thread to obtain subscription messages. You need to invoke OpenHiva::SpinOnce to obtain subscription messages.

#include <thread>
#include <unistd.h>
#include "open/subscriber.h"
#include "open/init.h"
#include "open/node.h"
#include "std_msgs/include/StringMessage.h"

// Users define callback functions based on their own service logic to process subscription messages.
void ChatterCallback(const Hiva::StdMgs::StringMessage msg)
{
    HIVA_WARN("I heard    :%s", msg.stringData.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
    const std::string groupName = testGroup.groupName;   // The value of groupName must be the same as that of ThreadGroup.groupName in the Init interface.
    const std::string topicName = "/chatter";           // Name of the subscribed topic, which must be the same as that of the topic published by the publisher.
    // Define a thread group.
    std::vector<OpenHiva::ScheduleGroup> threadGroup;
    OpenHiva::ScheduleGroup testGroup;
    testGroup.groupName =  "charter_listener";                       // Thread group name, which must be unique in each process.
    testGroup.scheduleType = OpenHiva::ScheduleType::UNBIND_AICPU;  // Indicates whether the working thread is deterministic.
    threadGroup.push_back(testGroup);
    // Invoke the resource initialization interface.
    OpenHiva::Init(argc, argv, threadGroup); 
    HIVA_ERROR("Listener init ok!");

    // 2. Subscribe to messages.
    // Construct the Node object.
    OpenHiva::Node n("/listener");
    // Construct TopicOptions.
    OpenHiva::TopicOptions topicOps;
    topicOps.BuildGroupName(groupName)
            .BuildMessageTraits<Hiva::StdMsgs::StringMessage>()
            .BuildShmOptions(maxMsgSize, blockNum)
            .BuildQueueOptions(queueSize, overwrite, queueFCFlag, queueTTL)
            .BuildTopicName(topicName);
    // Invoke the Subscriber interface through the NodeHandle object to subscribe to the topic. The Subscriber object is returned.
    std::function<void(Hiva::StdMgs::StringMessage)> callBack = ChatterCallback;
    std::shared_ptr<OpenHiva::Subscriber> sub = n.CreateSubscriber(topicName, callBack, topicOps);
    // Check whether the Subscriber object is successfully constructed. If the operation fails, invoke the Shutdown function to release resources and exit.
    if ((sub == nullptr) || !sub->Ready()) {
        HIVA_ERROR("sample_listener: sub.Ok() is false!");
        OpenHiva::Shutdown();    
        return 0;
}  
    // 3. Release resources.
    // 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()) {  
       OpenHiva::SpinOnce(testGroup.groupName);   // In USER_DEFINED mode, SpinOnce is invoked. In non-USER_DEFINED mode, SpinOnce does not need to be invoked.
        sleep(1);
        std::cout<<"I'm listerner"<<std::endl;
    }
    // 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;
}