freyo / laravel-queue-cmq

src/Queue/Driver/Queue.php

<?php

namespace Freyo\LaravelQueueCMQ\Queue\Driver;

class Queue
{
    private $queue_name;
    private $cmq_client;
    private $encoding;

    public function __construct($queue_name, CMQClient $cmq_client, $encoding = false)
    {
        $this->queue_name = $queue_name;
        $this->cmq_client = $cmq_client;
        $this->encoding = $encoding;
    }

    /**
     * @return mixed
     */
    public function getQueueName()
    {
        return $this->queue_name;
    }

    /* 设置是否对消息体进行base64编码

        @type encoding: bool
        @param encoding: 是否对消息体进行base64编码

The doc comment 是否对消息体进行base64编码 at position 0 could not be parsed: Unknown type name '是否对消息体进行base64编码' at position 0 in 是否对消息体进行base64编码.

    */
    public function set_encoding($encoding)
    {
        $this->encoding = $encoding;
    }

    /* 创建队列

        @type queue_meta: QueueMeta object
        @param queue_meta: QueueMeta对象，设置队列的属性

The doc comment QueueMeta对象，设置队列的属性 at position 0 could not be parsed: Unknown type name 'QueueMeta对象，设置队列的属性' at position 0 in QueueMeta对象，设置队列的属性.

    */
    public function create($queue_meta)
    {
        $params = [
            'queueName'           => $this->queue_name,
            'pollingWaitSeconds'  => $queue_meta->pollingWaitSeconds,
            'visibilityTimeout'   => $queue_meta->visibilityTimeout,
            'maxMsgSize'          => $queue_meta->maxMsgSize,
            'msgRetentionSeconds' => $queue_meta->msgRetentionSeconds,
            'rewindSeconds'       => $queue_meta->rewindSeconds,
        ];
        if ($queue_meta->maxMsgHeapNum > 0) {
            $params['maxMsgHeapNum'] = $queue_meta->maxMsgHeapNum;
        }
        $this->cmq_client->create_queue($params);
    }

    /* 获取队列属性

        @rtype: QueueMeta object
        @return 队列的属性

The doc comment 队列的属性 at position 0 could not be parsed: Unknown type name '队列的属性' at position 0 in 队列的属性.

    */
    public function get_attributes()
    {
        $params = [
            'queueName' => $this->queue_name,
        ];
        $resp = $this->cmq_client->get_queue_attributes($params);
        $queue_meta = new QueueMeta();
        $queue_meta->queueName = $this->queue_name;
        $this->__resp2meta__($queue_meta, $resp);

        return $queue_meta;
    }

    /* 设置队列属性

        @type queue_meta: QueueMeta object
        @param queue_meta: QueueMeta对象，设置队列的属性

The doc comment QueueMeta对象，设置队列的属性 at position 0 could not be parsed: Unknown type name 'QueueMeta对象，设置队列的属性' at position 0 in QueueMeta对象，设置队列的属性.

    */

    protected function __resp2meta__($queue_meta, $resp)
    {
        if (isset($resp['queueName'])) {
            $queue_meta->queueName = $resp['queueName'];
        }
        if (isset($resp['maxMsgHeapNum'])) {
            $queue_meta->maxMsgHeapNum = $resp['maxMsgHeapNum'];
        }
        if (isset($resp['pollingWaitSeconds'])) {
            $queue_meta->pollingWaitSeconds = $resp['pollingWaitSeconds'];
        }
        if (isset($resp['visibilityTimeout'])) {
            $queue_meta->visibilityTimeout = $resp['visibilityTimeout'];
        }
        if (isset($resp['maxMsgSize'])) {
            $queue_meta->maxMsgSize = $resp['maxMsgSize'];
        }
        if (isset($resp['msgRetentionSeconds'])) {
            $queue_meta->msgRetentionSeconds = $resp['msgRetentionSeconds'];
        }
        if (isset($resp['createTime'])) {
            $queue_meta->createTime = $resp['createTime'];
        }
        if (isset($resp['lastModifyTime'])) {
            $queue_meta->lastModifyTime = $resp['lastModifyTime'];
        }
        if (isset($resp['activeMsgNum'])) {
            $queue_meta->activeMsgNum = $resp['activeMsgNum'];
        }
        if (isset($resp['rewindSeconds'])) {
            $queue_meta->rewindSeconds = $resp['rewindSeconds'];
        }
        if (isset($resp['inactiveMsgNum'])) {
            $queue_meta->inactiveMsgNum = $resp['inactiveMsgNum'];
        }
        if (isset($resp['rewindmsgNum'])) {
            $queue_meta->rewindmsgNum = $resp['rewindmsgNum'];
        }
        if (isset($resp['minMsgTime'])) {
            $queue_meta->minMsgTime = $resp['minMsgTime'];
        }
        if (isset($resp['delayMsgNum'])) {
            $queue_meta->delayMsgNum = $resp['delayMsgNum'];
        }
    }

    public function set_attributes($queue_meta)
    {
        $params = [
            'queueName'           => $this->queue_name,
            'pollingWaitSeconds'  => $queue_meta->pollingWaitSeconds,
            'visibilityTimeout'   => $queue_meta->visibilityTimeout,
            'maxMsgSize'          => $queue_meta->maxMsgSize,
            'msgRetentionSeconds' => $queue_meta->msgRetentionSeconds,
            'rewindSeconds'       => $queue_meta->rewindSeconds,
        ];
        if ($queue_meta->maxMsgHeapNum > 0) {
            $params['maxMsgHeapNum'] = $queue_meta->maxMsgHeapNum;
        }

        $this->cmq_client->set_queue_attributes($params);
    }

    /* 删除队列

    */

    public function rewindQueue($backTrackingTime)
    {
        $params = [
            'queueName'        => $this->queue_name,
            'startConsumeTime' => $backTrackingTime,
        ];
        $this->cmq_client->rewindQueue($params);
    }

    /* 发送消息

        @type message: Message object
        @param message: 发送的Message object

The doc comment 发送的Message at position 0 could not be parsed: Unknown type name '发送的Message' at position 0 in 发送的Message.

        @rtype: Message object
        @return 消息发送成功的返回属性，包含MessageId

The doc comment 消息发送成功的返 at position 0 could not be parsed: Unknown type name '消息发送成功的返回属性，包含MessageId' at position 0 in 消息发送成功的返回属性，包含MessageId.

    */

    public function delete()
    {
        $params = ['queueName' => $this->queue_name];
        $this->cmq_client->delete_queue($params);
    }

    /* 批量发送消息

       @type messages: list of Message object
       @param messages: 发送的Message object list

The doc comment 发送的Message at position 0 could not be parsed: Unknown type name '发送的Message' at position 0 in 发送的Message.

       @rtype: list of Message object
       @return 多条消息发送成功的返回属性，包含MessageId

The doc comment 多条消息发送成功 at position 0 could not be parsed: Unknown type name '多条消息发送成功的返回属性，包含MessageId' at position 0 in 多条消息发送成功的返回属性，包含MessageId.

    */

    public function send_message($message, $delayTime = 0)
    {
        if ($this->encoding) {
            $msgBody = base64_encode($message->msgBody);
        } else {
            $msgBody = $message->msgBody;
        }
        $params = [
            'queueName'    => $this->queue_name,
            'msgBody'      => $msgBody,
            'delaySeconds' => $delayTime,
        ];
        $msgId = $this->cmq_client->send_message($params);
        $retmsg = new Message();
        $retmsg->msgId = $msgId;

        return $retmsg;
    }

    /* 消费消息

        @type polling_wait_seconds: int
        @param polling_wait_seconds: 本次请求的长轮询时间，单位：秒

The doc comment 本次请求的长轮询时间，单位：秒 at position 0 could not be parsed: Unknown type name '本次请求的长轮询时间，单位：秒' at position 0 in 本次请求的长轮询时间，单位：秒.

        @rtype: Message object
        @return Message object中包含基本属性、临时句柄
    */

    public function batch_send_message($messages, $delayTime = 0)
    {
        $params = [
            'queueName'    => $this->queue_name,
            'delaySeconds' => $delayTime,
        ];
        $n = 1;
        foreach ($messages as $message) {
            $key = 'msgBody.'.$n;
            if ($this->encoding) {
                $params[$key] = base64_encode($message->msgBody);
            } else {
                $params[$key] = $message->msgBody;
            }
            $n += 1;
        }
        $msgList = $this->cmq_client->batch_send_message($params);
        $retMessageList = [];
        foreach ($msgList as $msg) {
            $retmsg = new Message();
            $retmsg->msgId = $msg['msgId'];
            $retMessageList[] = $retmsg;
        }

        return $retMessageList;

The expression return $retMessageList returns the type Freyo\LaravelQueueCMQ\Queue\Driver\Message[]|array which is incompatible with the documented return type Freyo\LaravelQueueCMQ\Queue\Driver\Message.

    }

    /* 批量消费消息

        @type num_of_msg: int
        @param num_of_msg: 本次请求最多获取的消息条数

The doc comment 本次请求最多获取的消息条数 at position 0 could not be parsed: Unknown type name '本次请求最多获取的消息条数' at position 0 in 本次请求最多获取的消息条数.

        @type polling_wait_seconds: int
        @param polling_wait_seconds: 本次请求的长轮询时间，单位：秒

        @rtype: list of Message object
        @return 多条消息的属性，包含消息的基本属性、临时句柄

The doc comment 多条消息的属性， at position 0 could not be parsed: Unknown type name '多条消息的属性，包含消息的基本属性、临时句柄' at position 0 in 多条消息的属性，包含消息的基本属性、临时句柄.

    */

    public function receive_message($polling_wait_seconds = null)
    {
        $params = ['queueName' => $this->queue_name];
        if ($polling_wait_seconds !== null) {
            $params['UserpollingWaitSeconds'] = $polling_wait_seconds;
            $params['pollingWaitSeconds'] = $polling_wait_seconds;
        } else {
            $params['UserpollingWaitSeconds'] = 30;
        }
        $resp = $this->cmq_client->receive_message($params);
        $msg = new Message();
        if ($this->encoding) {
            $msg->msgBody = base64_decode($resp['msgBody']);
        } else {
            $msg->msgBody = $resp['msgBody'];
        }
        $msg->msgId = $resp['msgId'];
        $msg->receiptHandle = $resp['receiptHandle'];
        $msg->enqueueTime = $resp['enqueueTime'];
        $msg->nextVisibleTime = $resp['nextVisibleTime'];

The property nextVisibleTime does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

        $msg->dequeueCount = $resp['dequeueCount'];

The property dequeueCount does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

        $msg->firstDequeueTime = $resp['firstDequeueTime'];

The property firstDequeueTime does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

        return $msg;
    }

    /* 删除消息

        @type receipt_handle: string
        @param receipt_handle: 最近一次操作该消息返回的临时句柄

The doc comment 最近一次操作该消息返回的临时句柄 at position 0 could not be parsed: Unknown type name '最近一次操作该消息返回的临时句柄' at position 0 in 最近一次操作该消息返回的临时句柄.

    */

    public function batch_receive_message($num_of_msg, $polling_wait_seconds = null)
    {
        $params = ['queueName' => $this->queue_name, 'numOfMsg' => $num_of_msg];
        if ($polling_wait_seconds != null) {
            $params['UserpollingWaitSeconds'] = $polling_wait_seconds;
            $params['pollingWaitSeconds'] = $polling_wait_seconds;
        } else {
            $params['UserpollingWaitSeconds'] = 30;
        }
        $msgInfoList = $this->cmq_client->batch_receive_message($params);
        $retMessageList = [];
        foreach ($msgInfoList as $msg) {
            $retmsg = new Message();
            if ($this->encoding) {
                $retmsg->msgBody = base64_decode($msg['msgBody']);
            } else {
                $retmsg->msgBody = $msg['msgBody'];
            }
            $retmsg->msgId = $msg['msgId'];
            $retmsg->receiptHandle = $msg['receiptHandle'];
            $retmsg->enqueueTime = $msg['enqueueTime'];
            $retmsg->nextVisibleTime = $msg['nextVisibleTime'];

The property nextVisibleTime does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

            $retmsg->dequeueCount = $msg['dequeueCount'];

The property dequeueCount does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

            $retmsg->firstDequeueTime = $msg['firstDequeueTime'];

The property firstDequeueTime does not exist. Although not strictly required by PHP, it is generally a best practice to declare properties explicitly.

            $retMessageList[] = $retmsg;
        }

        return $retMessageList;
    }

    /* 批量删除消息

        @type receipt_handle_list: list
        @param receipt_handle_list: batch_receive_message返回的多条消息的临时句柄

The doc comment batch_receive_message返 at position 0 could not be parsed: Unknown type name 'batch_receive_message返回的多条消息的临时句柄' at position 0 in batch_receive_message返回的多条消息的临时句柄.

    */

    public function delete_message($receipt_handle)
    {
        $params = ['queueName' => $this->queue_name, 'receiptHandle' => $receipt_handle];
        $this->cmq_client->delete_message($params);
    }

    public function batch_delete_message($receipt_handle_list)
    {
        $params = ['queueName' => $this->queue_name];
        $n = 1;
        foreach ($receipt_handle_list as $receipt_handle) {
            $key = 'receiptHandle.'.$n;
            $params[$key] = $receipt_handle;
            $n += 1;
        }
        $this->cmq_client->batch_delete_message($params);
    }
}