Message Subscription

Principles

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

Figure 1 Process of invoking the BufferMessage subscription interface

1. Initialize resources.

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

2. Subscribe to messages.
   1. Before subscribing to a topic, you must create a node handle OpenHiva::Node n.
   2. Use the created node handle n to invoke the CreateSubscriber interface to subscribe to the specified topic. After the CreateSubscriber interface is invoked, the Subscriber object is returned.
      

      Pay attention to the following when setting the input parameters of the CreateSubscriber interface:

      • The input parameter topicName of the CreateSubscriber interface of the subscriber must be the same as that of the CreatePublisher interface of the publisher.
      • The input parameter groupName of the CreateSubscriber interface must be the same as that set in the OpenHiva::Init interface.

      The message subscription process over the CreateSubscriber interface is as follows:

      1. Register the callback function of a specified topic with the working thread to process messages received in the subscription queue.
      2. Create a subscription queue and register subscriber information such as the queue ID and topic with the DataMaster.
      3. After receiving the registration message, OpenHiva (the DataMaster process) searches for the IDs of all subscription queues that subscribe to the topic and binds the IDs to the publish queue IDs.
      4. When a message is published to a publishing queue, the queue scheduler transfers the message from the publishing queue to the subscription queue, sends an event to the working thread of the subscription end, activates the worker thread, and invokes the callback function to process the message.
   3. Invoke the Ready interface of the Subscriber object to check whether the subscriber is successfully created.
3. Release resources.

   The process is similar to the process of Common Messages > Message Subscription > 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 subscribing to a BufferMessage message in a basic scenario, 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 <string>
#include <iostream>
#include <memory>
#include <unistd.h>
#include "open/subscriber.h"
#include "open/init.h"
#include "open/node.h"
#include "open/buffer_message.h"
#include "securec.h"
uint32_t cnt = 0;

void Usage()
{
    std::cout << "Usage: pmupload test_create_buffer_sub nodeName topicName groupName bindType maxMsgSize queueSize blockNum overwrite queueFCFlag queueTTL transport" << std::endl;
    std::cout << "if argc==2, pub will use default value: pmupload test_create_buffer_sub testNodeBufferSub testTopic testGroupBufferSub 0 102400 10 10 1 0 0 1" << 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;
}

// Users define callback functions based on their own service logic to process subscription messages.
void ChatterCallback(const OpenHiva::BufferMessage &msg)
{
    void *currBlockPtr = nullptr;
    size_t dataSize = 0U;
    OpenHiva::HivaBuffer &HivaBuffer = const_cast<OpenHiva::HivaBuffer &>(msg.data);
    uint32_t ret = HivaBuffer.GetBuff(currBlockPtr, dataSize);        // Obtain the memory addresses dataPtr and dataLen.
    if (ret != Hiva::HIVA_SUCCESS) {
        HIVA_ERROR("GetBuff failed and return %u", ret);
        return;
    }
    uint32_t dataLen = *reinterpret_cast<uint32_t *>(currBlockPtr);
    std::string str(reinterpret_cast<char *>(currBlockPtr) + 4, dataLen);
    HIVA_WARN("I heard    :%s", str.c_str());
    cnt++;
    return;
}

int main(int argc, char **argv)
{
    // 1. Initialize resources.
    std::string nodeName = "testNodeSub";         
    std::string topicName = "testTopic";           // Name of the subscribed topic, which must be the same as that of the topic published by the publisher.
    std::string groupName = "testGroupSub";        // The value of groupName must be the same as that of ThreadGroup.groupName in the Init interface.
    OpenHiva::ScheduleType scheType = OpenHiva::ScheduleType(0);
    uint32_t maxMsgSize = 100 * 1024U;
    uint32_t queueSize = 10U;
    uint32_t blockNum = 10U;
    bool overwrite = false;
    bool queueFCFlag = false;
    uint16_t queueTTL = 100U;
    std::string transport;
    int argNum = 12;
    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);
        transport = argv[11];
    }
    // Define a thread group.
    std::vector<OpenHiva::ScheduleGroup> scheGrpVec;
    OpenHiva::ScheduleGroup scheGrp;
    scheGrp.groupName = groupName;                  // Thread group name, which must be unique in each process.
    scheGrp.scheduleType = scheType;                // Indicates whether the working thread is deterministic.
    scheGrpVec.push_back(scheGrp);
    // Invoke the resource initialization interface.
    OpenHiva::Init(argc, argv, scheGrpVec);
    HIVA_EVENT("subscriber init ok!");

    // 2. Subscribe to 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 Subscriber interface through the NodeHandle object to subscribe to the topic. The Subscriber object is returned.
    std::function<void(OpenHiva::BufferMessage)> callBack = ChatterCallback;
    std::shared_ptr<OpenHiva::Subscriber> sub = node.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("create subscriber failed");
        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(groupName)   // In USER_DEFINED mode, SpinOnce is invoked. In non-USER_DEFINED mode, SpinOnce does not need to be invoked.
        sleep(1);
    }
    // This function is invoked in blocking mode to prevent the process from proactively exiting.
    OpenHiva::WaitForShutdown();
    return 0;
}