Send and receive ordered messages

Background information

Ordered messages are classified into the following types:

• Globally ordered messages: All messages of a specified topic are published and consumed in first-in-first-out (FIFO) order.

• Partitionally ordered messages: All messages of a specified topic are distributed to different partitions by using sharding keys. The messages in each partition are published and consumed in FIFO order. A sharding key is a key field that is used for ordered messages to identify different partitions. A sharding key is different from the key of a normal message.

For more information, see Ordered messages.

Prerequisites

Make sure that the following operations are performed:

• The SDK for C or C++ is downloaded. For more information, see Release notes for the TCP client SDK for C++.

• The environment is prepared. For more information, see Prepare the environment (v1.x.x).

• The resources that you want to specify in the code are created in the ApsaraMQ for RocketMQ console. The resources include instances, topics, and consumer groups. For more information, see Create resources.

• The AccessKey pair of your Alibaba Cloud account is obtained. For more information, see Create an AccessKey.

Send ordered messages

The following sample code provides an example on how to send ordered messages by using the TCP client SDK for C or C++:

#include "ONSFactory.h"
#include "ONSClientException.h"
#include <iostream>
using namespace ons;

int main()
{
    // The parameter that is required to create and use a producer. 
    ONSFactoryProperty factoryInfo;
    // The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
    // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX"); 
    // The topic that you created in the ApsaraMQ for RocketMQ console. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
    // The message content. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
    // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
    // The AccessKey ID that is used for authentication. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
		// The AccessKey secret that is used for authentication. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

    // Create the producer. 
    OrderProducer *pProducer = ONSFactory::getInstance()->createOrderProducer(factoryInfo);


    // Before you send the message, call the start() method only once to start the producer. 
    pProducer->start();

    Message msg(
                //Message Topic
                factoryInfo.getPublishTopics(),
                //Message Tag
                "TagA",
                //Message Key. Set a key to uniquely identify the message. We recommend that the key be globally unique so that if a message cannot be received, you can query it in the ApsaraMQ for RocketMQ console using the key and resend it. Note: Not setting a key does not affect message sending or receiving.
                "ORDERID_100",
                //Message Body
                factoryInfo.getMessageContent()
    );
    // The key field that is used to identify partitions for partitionally ordered messages. 
    // This field can be set to a non-empty string for globally ordered messages. 
    std::string shardingKey = "abc";  
    // Messages that have the same sharding key are sent in order. 
    try
    {
        // Send the message. If no exception is thrown, the message is sent. 
        SendResultONS sendResult = pProducer->send(msg, shardingKey);
    std::cout << "send success" << std::endl;
    }
    catch(ONSClientException & e)
    {
        // Specify the logic to handle errors. 
    }
    // Before you exit your application, destroy the producer. If you do not destroy the producer, issues such as memory leaks may occur. 
    pProducer->shutdown();

    return 0;
}           

Subscribe to ordered messages

The following sample code provides an example on how to subscribe to ordered messages by using the TCP client SDK for C or C++:

#include "ONSFactory.h"
using namespace std;
using namespace ons;

// Create the consumer instance. 
//After the push consumer pulls the message, the push consumer calls the consumeMessage function of the instance. 
class ONSCLIENT_API MyMsgListener : public MessageOrderListener
{
public:
    MyMsgListener()
    {
    }
    virtual ~MyMsgListener()
    {
    }

    virtual OrderAction consume(Message &message, ConsumeOrderContext &context)
    {
        // Consume messages based on your business requirements. 
        return Success; //CONSUME_SUCCESS;
    }
};


int main(int argc, char* argv[])
{
    // The parameter that is required to create and use the consumer. 
    ONSFactoryProperty factoryInfo;
    // The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::ConsumerId, "XXX");
    // The topic that you created in the ApsaraMQ for RocketMQ console. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
    // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
    // The AccessKey ID that is used for authentication. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
		// The AccessKey secret that is used for authentication. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
    // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
    factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX");


    // Create the consumer. 
    OrderConsumer* orderConsumer = ONSFactory::getInstance()->createOrderConsumer(factoryInfo);
    MyMsgListener  msglistener;
    // The message topic and tag to which the consumer subscribes. 
    orderConsumer->subscribe(factoryInfo.getPublishTopics(), "*",&msglistener );

    // Register the instance to listen to messages. After the consumer pulls the messages, the consumer calls the consumeMessage function of the message listening class. 

    //Start the consumer. 
    orderConsumer->start();

    for(volatile int i = 0; i < 1000000000; ++i) {
        //wait
    }

    // Destroy the consumer. Before you exit the application, destroy the consumer. Otherwise, issues such as memory leaks may occur. 
    orderConsumer->shutdown();

   return 0;
}