ApsaraMQ for RocketMQ:Send and receive transactional messages

Step 1: Implement the transaction executor

The transaction executor runs immediately after the half message is sent. It executes the local transaction logic and returns one of three statuses:

TransactionStatus.Unknow is the actual enum spelling in the SDK (without the trailing *n*).

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Runtime.InteropServices;
using ons;

namespace ons
{
    public class MyLocalTransactionExecuter : LocalTransactionExecuter
    {
        public MyLocalTransactionExecuter()
        {
        }

        ~MyLocalTransactionExecuter()
        {
        }

        public override TransactionStatus execute(Message value)
        {
            Console.WriteLine("execute topic: {0}, tag:{1}, key:{2}, msgId:{3},msgbody:{4}, userProperty:{5}",
                value.getTopic(), value.getTag(), value.getKey(), value.getMsgID(), value.getBody(), value.getUserProperties("VincentNoUser"));

            // Two messages can have the same message body but different IDs.
            // You cannot query the ID of the current message in the ApsaraMQ for RocketMQ console.
            // Use the message ID or a CRC32/MD5 hash of the message body
            // to implement idempotent processing and prevent duplicate messages.
            string msgId = value.getMsgID();

            TransactionStatus transactionStatus = TransactionStatus.Unknow;
            try {
                boolean isCommit = The execution result of the local transaction;
                if (isCommit) {
                    // Local transaction succeeded -- commit the message.
                    transactionStatus = TransactionStatus.CommitTransaction;
                } else {
                    // Local transaction failed -- roll back the message.
                    transactionStatus = TransactionStatus.RollbackTransaction;
                }
            } catch (Exception e) {
                // Handle exceptions from the local transaction.
                // Returning Unknow triggers the transaction checker later.
            }
            return transactionStatus;
        }
    }
}

Step 2: Implement the transaction checker

The transaction checker handles status-check requests from the broker. If the broker does not receive a commit or rollback for a half message -- for example, the application exits or the executor returns TransactionStatus.Unknow -- it periodically queries a producer instance in the producer cluster.

The check method must:

1. Determine whether the local transaction corresponding to the half message was committed or rolled back.

2. Return the appropriate TransactionStatus to the broker.

public class MyLocalTransactionChecker : LocalTransactionChecker
{
    public MyLocalTransactionChecker()
    {
    }

    ~MyLocalTransactionChecker()
    {
    }

    public override TransactionStatus check(Message value)
    {
        Console.WriteLine("check topic: {0}, tag:{1}, key:{2}, msgId:{3},msgbody:{4}, userProperty:{5}",
            value.getTopic(), value.getTag(), value.getKey(), value.getMsgID(), value.getBody(), value.getUserProperties("VincentNoUser"));

        // Two messages can have the same message body but different IDs.
        // You cannot query the ID of the current message in the ApsaraMQ for RocketMQ console.
        // Use the message ID or a CRC32/MD5 hash of the message body
        // to look up the local transaction result.
        // If your business logic is idempotent, the deduplication check is optional.
        string msgId = value.getMsgID();

        TransactionStatus transactionStatus = TransactionStatus.Unknow;
        try {
            boolean isCommit = The execution result of the local transaction;
            if (isCommit) {
                // Local transaction succeeded -- commit the message.
                transactionStatus = TransactionStatus.CommitTransaction;
            } else {
                // Local transaction failed -- roll back the message.
                transactionStatus = TransactionStatus.RollbackTransaction;
            }
        } catch (Exception e) {
            // Handle exceptions. Returning Unknow keeps the half message
            // pending and the broker will check again later.
        }
        return transactionStatus;
    }
}

Step 3: Send a half message and run the transaction

Create a TransactionProducer with the transaction checker, then send a half message to trigger the transaction executor.

Replace the following placeholders with your actual values:

class onscsharp
{
    static void Main(string[] args)
    {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();

        // TCP endpoint. Find this value in the TCP Endpoint section
        // of the Instance Details page in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<tcp-endpoint>");

        // Topic created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<topic-name>");

        // Message content.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.MsgContent, "<message-body>");

        // Make sure the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID
        // and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

        // Create the transaction producer with the transaction checker.
        LocalTransactionChecker myChecker = new MyLocalTransactionChecker();
        TransactionProducer pProducer = ONSFactory.getInstance().createTransactionProducer(factoryInfo, myChecker);

        // Call start() once before sending messages.
        // After the producer starts, you can send messages concurrently from multiple threads.
        pProducer.start();

        Message msg = new Message(
            factoryInfo.getPublishTopics(),   // Topic
            "TagA",                           // Tag for message filtering
            factoryInfo.getMessageContent()   // Message body
        );

        // Set a message key for tracing. Use a business identifier
        // (for example, an order ID) that is globally unique when possible.
        // You can use this key to look up the message in the ApsaraMQ for RocketMQ console.
        msg.setKey("ORDERID_100");

        // Send the half message and run the local transaction.
        try
        {
            LocalTransactionExecuter myExecuter = new MyLocalTransactionExecuter();
            SendResultONS sendResult = pProducer.send(msg, myExecuter);
        }
        catch (ONSClientException e)
        {
            Console.WriteLine("\nexception of sendmsg:{0}", e.what());
        }

        // Shut down the producer before the application exits to release resources.
        // A producer cannot be restarted after shutdown.
        pProducer.shutdown();
    }
}