ApsaraMQ for RocketMQ:Send and consume scheduled and delayed messages by using the PHP HTTP SDK

<?php

require "vendor/autoload.php";

use MQ\Model\TopicMessage;
use MQ\MQClient;

class ProducerTest
{
    private $client;
    private $producer;

    public function __construct()
    {
        $this->client = new MQClient(
            // HTTP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console
            "<HTTP_ENDPOINT>",
            // AccessKey ID for authentication
            getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            // AccessKey secret for authentication
            getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
        );

        $topic = "<TOPIC>";
        // Instance ID. If the instance has no namespace, set to null or "".
        $instanceId = "<INSTANCE_ID>";

        $this->producer = $this->client->getProducer($instanceId, $topic);
    }

    public function run()
    {
        try
        {
            for ($i = 1; $i <= 4; $i++)
            {
                $publishMessage = new TopicMessage(
                    "hello mq! " // Message body
                );
                // Custom message property
                $publishMessage->putProperty("a", $i);
                // Message key for tracing
                $publishMessage->setMessageKey("MessageKey");

                // Deliver the message 10 seconds from now.
                // For a scheduled message, set this to the target delivery time
                // as a millisecond-level Unix timestamp.
                $publishMessage->setStartDeliverTime(time() * 1000 + 10 * 1000);

                $result = $this->producer->publishMessage($publishMessage);

                print "Send mq message success. msgId is:" . $result->getMessageId()
                    . ", bodyMD5 is:" . $result->getMessageBodyMD5() . "\n";
            }
        } catch (\Exception $e) {
            print_r($e->getMessage() . "\n");
        }
    }
}

$instance = new ProducerTest();
$instance->run();

?>

<?php

use MQ\MQClient;

require "vendor/autoload.php";

class ConsumerTest
{
    private $client;
    private $consumer;

    public function __construct()
    {
        $this->client = new MQClient(
            // HTTP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console
            "<HTTP_ENDPOINT>",
            // AccessKey ID for authentication
            getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            // AccessKey secret for authentication
            getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
        );

        $topic = "<TOPIC>";
        // Consumer group ID created in the ApsaraMQ for RocketMQ console
        $groupId = "<GROUP_ID>";
        // Instance ID. If the instance has no namespace, set to null or "".
        $instanceId = "<INSTANCE_ID>";

        $this->consumer = $this->client->getConsumer($instanceId, $topic, $groupId);
    }

    public function ackMessages($receiptHandles)
    {
        try {
            $this->consumer->ackMessage($receiptHandles);
        } catch (\Exception $e) {
            if ($e instanceof MQ\Exception\AckMessageException) {
                // ACK may fail if the receipt handle has expired
                printf("Ack Error, RequestId:%s\n", $e->getRequestId());
                foreach ($e->getAckMessageErrorItems() as $errorItem) {
                    printf("\tReceiptHandle:%s, ErrorCode:%s, ErrorMsg:%s\n",
                        $errorItem->getReceiptHandle(),
                        $errorItem->getErrorCode(),
                        $errorItem->getErrorMessage());
                }
            }
        }
    }

    public function run()
    {
        // Consume messages in a loop. For production use, run multiple threads
        // to consume messages concurrently.
        while (True) {
            try {
                // Long polling: if no message is available, the request waits
                // on the broker until a message arrives or the polling period ends.
                $messages = $this->consumer->consumeMessage(
                    3, // Max messages per request (up to 16)
                    3  // Long polling timeout in seconds (up to 30)
                );
            } catch (\MQ\Exception\MessageResolveException $e) {
                // Thrown when some messages contain invalid characters and cannot be parsed
                $messages = $e->getPartialResult()->getMessages();
                $failMessages = $e->getPartialResult()->getFailResolveMessages();

                $receiptHandles = array();
                foreach ($messages as $message) {
                    $receiptHandles[] = $message->getReceiptHandle();
                    printf("MsgID %s\n", $message->getMessageId());
                }
                foreach ($failMessages as $failMessage) {
                    $receiptHandles[] = $failMessage->getReceiptHandle();
                    printf("Fail To Resolve Message. MsgID %s\n", $failMessage->getMessageId());
                }
                $this->ackMessages($receiptHandles);
                continue;
            } catch (\Exception $e) {
                if ($e instanceof MQ\Exception\MessageNotExistException) {
                    // No messages available. Long polling continues on the next iteration.
                    printf("No message, continue long polling! RequestId:%s\n", $e->getRequestId());
                    continue;
                }

                print_r($e->getMessage() . "\n");

                sleep(3);
                continue;
            }

            print "consume finish, messages:\n";

            $receiptHandles = array();
            foreach ($messages as $message) {
                $receiptHandles[] = $message->getReceiptHandle();

                printf("MessageID:%s TAG:%s BODY:%s \nPublishTime:%d, FirstConsumeTime:%d, \nConsumedTimes:%d, NextConsumeTime:%d, MessageKey:%s\n",
                    $message->getMessageId(), $message->getMessageTag(), $message->getMessageBody(),
                    $message->getPublishTime(), $message->getFirstConsumeTime(),
                    $message->getConsumedTimes(), $message->getNextConsumeTime(),
                    $message->getMessageKey());
                print_r($message->getProperties());
            }

            // If the broker does not receive an ACK before the time specified
            // in getNextConsumeTime(), the message is redelivered.
            // Each delivery generates a new receipt handle.
            print_r($receiptHandles);
            $this->ackMessages($receiptHandles);
            print "ack finish\n";
        }

    }
}

$instance = new ConsumerTest();
$instance->run();

?>