mns/queue.py

#coding=utf-8
# Copyright (C) 2015, Alibaba Cloud Computing

#Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

#The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

#THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

import time
from .mns_client import MNSClient
from .mns_request import *
from .mns_exception import *

class Queue:
    def __init__(self, queue_name, mns_client, debug=False):
        self.queue_name = queue_name
        self.mns_client = mns_client
        self.set_encoding(True)
        self.debug = debug

    def set_debug(self, debug):
        self.debug = debug

    def set_encoding(self, encoding):
        """ 设置是否对消息体进行base64编码

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

    def create(self, queue_meta, req_info=None):
        """ 创建队列

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @rtype: string
            @return 新创建队列的URL

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = CreateQueueRequest(self.queue_name, queue_meta.visibility_timeout, queue_meta.maximum_message_size, queue_meta.message_retention_period, queue_meta.delay_seconds, queue_meta.polling_wait_seconds, queue_meta.logging_enabled)
        req.set_req_info(req_info)
        resp = CreateQueueResponse()
        self.mns_client.create_queue(req, resp)
        self.debuginfo(resp)
        return resp.queue_url

    def get_attributes(self, req_info=None):
        """ 获取队列属性

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @note: Exception
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = GetQueueAttributesRequest(self.queue_name)
        req.set_req_info(req_info)
        resp = GetQueueAttributesResponse()
        self.mns_client.get_queue_attributes(req, resp)
        queue_meta = QueueMeta()
        self.__resp2meta__(queue_meta, resp)
        self.debuginfo(resp)
        return queue_meta

    def set_attributes(self, queue_meta, req_info=None):
        """ 设置队列属性

            @type queue_meta: QueueMeta object
            @param queue_meta: 新设置的属性

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = SetQueueAttributesRequest(self.queue_name, queue_meta.visibility_timeout, queue_meta.maximum_message_size, queue_meta.message_retention_period, queue_meta.delay_seconds, queue_meta.polling_wait_seconds, queue_meta.logging_enabled)
        req.set_req_info(req_info)
        resp = SetQueueAttributesResponse()
        self.mns_client.set_queue_attributes(req, resp)
        self.debuginfo(resp)

    def delete(self, req_info=None):
        """ 删除队列

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @note: Exception
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = DeleteQueueRequest(self.queue_name)
        req.set_req_info(req_info)
        resp = DeleteQueueResponse()
        self.mns_client.delete_queue(req, resp)
        self.debuginfo(resp)

    def send_message(self, message, req_info=None):
        """ 发送消息

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

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

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = SendMessageRequest(self.queue_name, message.message_body, message.delay_seconds, message.priority, self.encoding)
        req.set_req_info(req_info)
        resp = SendMessageResponse()
        self.mns_client.send_message(req, resp)
        self.debuginfo(resp)
        return self.__send_resp2msg__(resp)

    def batch_send_message(self, messages, req_info=None):
        """批量发送消息
           
            @type messages: list of Message object
            @param messages: 发送的Message object list

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

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

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = BatchSendMessageRequest(self.queue_name, self.encoding)
        req.set_req_info(req_info)
        for msg in messages:
            req.add_message(msg.message_body, msg.delay_seconds, msg.priority)
        resp = BatchSendMessageResponse()
        self.mns_client.batch_send_message(req, resp)
        self.debuginfo(resp)
        return self.__batchsend_resp2msg__(resp)

    def peek_message(self, req_info=None):
        """ 查看消息 消息体为字节串

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @rtype: Message object
            @return: Message object中包含消息的基本属性

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = PeekMessageRequest(self.queue_name, self.encoding)
        req.set_req_info(req_info)
        resp = PeekMessageResponse()
        self.mns_client.peek_message(req, resp)
        self.debuginfo(resp)
        return self.__peek_resp2msg__(resp)

    def peek_message_with_str_body(self, req_info=None):
        """ 查看消息 消息体为字符串

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @rtype: Message object
            @return: Message object中包含消息的基本属性

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        message = self.peek_message(req_info)
        if self.encoding:
            message.message_body = message.message_body.decode("utf-8")

        return message

    def batch_peek_message(self, batch_size, req_info=None):
        """ 批量查看消息 消息体为字节串
            
            @type batch_size: int
            @param batch_size: 本次请求最多获取的消息条数

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

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

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = BatchPeekMessageRequest(self.queue_name, batch_size, self.encoding)
        req.set_req_info(req_info)
        resp = BatchPeekMessageResponse()
        self.mns_client.batch_peek_message(req, resp)
        self.debuginfo(resp)
        return self.__batchpeek_resp2msg__(resp)

    def batch_peek_message_with_str_body(self, batch_size, req_info=None):
        """ 批量查看消息 消息体为字符串

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

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

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        msg_list = self.batch_peek_message(batch_size, req_info)
        if self.encoding:
            for message in msg_list:
                message.message_body = message.message_body.decode("utf-8")

        return msg_list

    def receive_message(self, wait_seconds = -1, req_info=None):
        """ 消费消息 消息体为字节串

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @rtype: Message object
            @return Message object中包含基本属性、下次可消费时间和临时句柄

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = ReceiveMessageRequest(self.queue_name, self.encoding, wait_seconds)
        req.set_req_info(req_info)
        resp = ReceiveMessageResponse()
        self.mns_client.receive_message(req, resp)
        self.debuginfo(resp)
        return self.__recv_resp2msg__(resp)

    def receive_message_with_str_body(self, wait_seconds = -1, req_info=None):
        """ 消费消息 消息体为字符串

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @rtype: Message object
            @return Message object中包含基本属性、下次可消费时间和临时句柄

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        message = self.receive_message(wait_seconds, req_info)
        if self.encoding:
            message.message_body = message.message_body.decode("utf-8")

        return message

    def batch_receive_message(self, batch_size, wait_seconds = -1, req_info=None):
        """ 批量消费消息 消息体为字节串

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

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

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

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = BatchReceiveMessageRequest(self.queue_name, batch_size, self.encoding, wait_seconds)
        req.set_req_info(req_info)
        resp = BatchReceiveMessageResponse()
        self.mns_client.batch_receive_message(req, resp)
        self.debuginfo(resp)
        return self.__batchrecv_resp2msg__(resp)

    def batch_receive_message_with_str_body(self, batch_size, wait_seconds = -1, req_info=None):
        """ 批量消费消息 消息体为字符串

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

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

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

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        msg_list = self.batch_receive_message(batch_size, wait_seconds, req_info)
        if self.encoding:
            for message in msg_list:
                message.message_body = message.message_body.decode("utf-8")

        return msg_list

    def delete_message(self, receipt_handle, req_info=None):
        """ 删除消息

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

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = DeleteMessageRequest(self.queue_name, receipt_handle)
        req.set_req_info(req_info)
        resp = DeleteMessageResponse()
        self.mns_client.delete_message(req, resp)
        self.debuginfo(resp)

    def batch_delete_message(self, receipt_handle_list, req_info=None):
        """批量删除消息
            
            @type receipt_handle_list: list
            @param receipt_handle_list: batch_receive_message返回的多条消息的临时句柄

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = BatchDeleteMessageRequest(self.queue_name, receipt_handle_list)
        req.set_req_info(req_info)
        resp = BatchDeleteMessageResponse()
        self.mns_client.batch_delete_message(req, resp)
        self.debuginfo(resp)

    def change_message_visibility(self, reciept_handle, visibility_timeout, req_info=None):
        """ 修改消息下次可消费时间

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

            @type visibility_timeout: int
            @param visibility_timeout: 消息下次可被消费时间为
                                       now+visibility_timeout, 单位：秒

            @type req_info: RequestInfo object
            @param req_info: 透传到MNS的请求信息

            @rtype: Message object
            @return: Message object包含临时句柄和下次可消费时间

            @note: Exception
            :: MNSClientParameterException  参数格式异常
            :: MNSClientNetworkException    网络异常
            :: MNSServerException           mns处理异常
        """
        req = ChangeMessageVisibilityRequest(self.queue_name, reciept_handle, visibility_timeout)
        req.set_req_info(req_info)
        resp = ChangeMessageVisibilityResponse()
        self.mns_client.change_message_visibility(req, resp)
        self.debuginfo(resp)
        return self.__changevis_resp2msg__(resp)

    def debuginfo(self, resp):
        if self.debug:
            print("===================DEBUG INFO===================")
            print("RequestId: %s" % resp.header["x-mns-request-id"])
            print("================================================")

    def __resp2meta__(self, queue_meta, resp):
        queue_meta.visibility_timeout = resp.visibility_timeout
        queue_meta.maximum_message_size = resp.maximum_message_size
        queue_meta.message_retention_period = resp.message_retention_period
        queue_meta.delay_seconds = resp.delay_seconds
        queue_meta.polling_wait_seconds = resp.polling_wait_seconds
        queue_meta.logging_enabled = resp.logging_enabled

        queue_meta.active_messages = resp.active_messages
        queue_meta.inactive_messages = resp.inactive_messages
        queue_meta.delay_messages = resp.delay_messages
        queue_meta.create_time = resp.create_time
        queue_meta.last_modify_time = resp.last_modify_time
        queue_meta.queue_name = resp.queue_name

    def __send_resp2msg__(self, resp):
        msg = Message()
        msg.message_id = resp.message_id
        msg.message_body_md5 = resp.message_body_md5
        msg.receipt_handle = resp.receipt_handle
        return msg

    def __batchsend_resp2msg__(self, resp):
        msg_list = []
        for entry in resp.message_list:
            msg = Message()
            msg.message_id = entry.message_id
            msg.message_body_md5 = entry.message_body_md5
            msg_list.append(msg)
        return msg_list

    def __peek_resp2msg__(self, resp):
        msg = Message()
        msg.message_id = resp.message_id
        msg.message_body_md5 = resp.message_body_md5
        msg.dequeue_count = resp.dequeue_count
        msg.enqueue_time = resp.enqueue_time
        msg.first_dequeue_time = resp.first_dequeue_time
        msg.message_body = resp.message_body
        msg.priority = resp.priority
        return msg

    def __batchpeek_resp2msg__(self, resp):
        msg_list = []
        for entry in resp.message_list:
            msg = Message()
            msg.message_id = entry.message_id
            msg.message_body_md5 = entry.message_body_md5
            msg.dequeue_count = entry.dequeue_count
            msg.enqueue_time = entry.enqueue_time
            msg.first_dequeue_time = entry.first_dequeue_time
            msg.message_body = entry.message_body
            msg.priority = entry.priority
            msg_list.append(msg)
        return msg_list

    def __recv_resp2msg__(self, resp):
        msg = self.__peek_resp2msg__(resp)
        msg.receipt_handle = resp.receipt_handle
        msg.next_visible_time = resp.next_visible_time
        return msg

    def __batchrecv_resp2msg__(self, resp):
        msg_list = []
        for entry in resp.message_list:
            msg = Message()
            msg.message_id = entry.message_id
            msg.message_body_md5 = entry.message_body_md5
            msg.dequeue_count = entry.dequeue_count
            msg.enqueue_time = entry.enqueue_time
            msg.first_dequeue_time = entry.first_dequeue_time
            msg.message_body = entry.message_body
            msg.priority = entry.priority
            msg.next_visible_time = entry.next_visible_time
            msg.receipt_handle = entry.receipt_handle
            msg_list.append(msg)
        return msg_list

    def __changevis_resp2msg__(self, resp):
        msg = Message()
        msg.receipt_handle = resp.receipt_handle
        msg.next_visible_time = resp.next_visible_time
        return msg


class QueueMeta:
    DEFAULT_VISIBILITY_TIMEOUT = -1
    DEFAULT_MAXIMUM_MESSAGE_SIZE = -1
    DEFAULT_MESSAGE_RETENTION_PERIOD = -1
    DEFAULT_DELAY_SECONDS = -1
    DEFAULT_POLLING_WAIT_SECONDS = -1

    def __init__(self, vis_timeout=None, max_msg_size=None, msg_ttl=None,
                 delay_sec=None, polling_wait_sec=None, logging_enabled=None):
        """ 队列属性
            @note: 设置属性
            :: visibility_timeout: message被receive后，持续不可消费的时间, 单位：秒
            :: maximum_message_size: message body的最大长度, 单位：Byte
            :: message_retention_period: message最长存活时间，单位：秒
            :: delay_seconds: 新message可消费的默认延迟时间，单位：秒
            :: polling_wait_seconds: receive message时，长轮询时间，单位：秒
            :: logging_enabled: 是否开启logging功能，如果开启MNS将该队列的日志推送到Account的logging bucket中
            
            @note: 非设置属性
            :: active_messages: 可消费消息数，近似值
            :: inactive_messages: 正在被消费的消息数，近似值
            :: delay_messages: 延迟消息数，近似值
            :: create_time: queue创建时间，单位：秒 
            :: last_modify_time: 修改queue属性的最近时间，单位：秒
            :: queue_name: 队列名称
        """
        self.visibility_timeout = QueueMeta.DEFAULT_VISIBILITY_TIMEOUT if vis_timeout is None else vis_timeout
        self.maximum_message_size = QueueMeta.DEFAULT_MAXIMUM_MESSAGE_SIZE if max_msg_size is None else max_msg_size
        self.message_retention_period = QueueMeta.DEFAULT_MESSAGE_RETENTION_PERIOD if msg_ttl is None else msg_ttl
        self.delay_seconds = QueueMeta.DEFAULT_DELAY_SECONDS if delay_sec is None else delay_sec
        self.polling_wait_seconds = QueueMeta.DEFAULT_POLLING_WAIT_SECONDS if polling_wait_sec is None else polling_wait_sec
        self.logging_enabled = logging_enabled

        self.active_messages = -1
        self.inactive_messages = -1
        self.delay_messages = -1
        self.create_time = -1
        self.last_modify_time = -1
        self.queue_name = ""

    def set_visibilitytimeout(self, visibility_timeout):
        self.visibility_timeout = visibility_timeout

    def set_maximum_message_size(self, maximum_message_size):
        self.maximum_message_size = maximum_message_size

    def set_message_retention_period(self, message_retention_period):
        self.message_retention_period = message_retention_period

    def set_delay_seconds(self, delay_seconds):
        self.delay_seconds = delay_seconds

    def set_polling_wait_seconds(self, polling_wait_seconds):
        self.polling_wait_seconds = polling_wait_seconds

    def set_logging_enabled(self, logging_enabled):
        self.logging_enabled = logging_enabled

    def __str__(self):
        meta_info = {"VisibilityTimeout": self.visibility_timeout,
                     "MaximumMessageSize": self.maximum_message_size,
                     "MessageRetentionPeriod": self.message_retention_period,
                     "DelaySeconds": self.delay_seconds,
                     "PollingWaitSeconds": self.polling_wait_seconds,
                     "ActiveMessages": self.active_messages,
                     "InactiveMessages": self.inactive_messages,
                     "DelayMessages": self.delay_messages,
                     "CreateTime": time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(self.create_time)),
                     "LastModifyTime": time.strftime("%Y/%m/%d %H:%M:%S", time.localtime(self.last_modify_time)),
                     "QueueName": self.queue_name,
                     "LoggingEnabled": self.logging_enabled}
        return "\n".join(["%s: %s" % (k.ljust(30), v) for k, v in meta_info.items()])


class Message:
    def __init__(self, message_body = None, delay_seconds = None, priority = None):
        """ 消息属性

            @note: send_message 指定属性
            :: message_body         消息体 
            :: delay_seconds        消息延迟时间
            :: priority             消息优先级

            @note: send_message 返回属性
            :: message_id           消息编号
            :: message_body_md5     消息体的MD5值

            @note: peek_message 返回属性(基本属性)
            :: message_body         消息体
            :: message_id           消息编号
            :: message_body_md5     消息体的MD5值
            :: dequeue_count        消息被消费的次数
            :: enqueue_time         消息发送到队列的时间，单位：毫秒
            :: first_dequeue_time   消息第一次被消费的时间，单位：毫秒

            @note: receive_message 返回属性，除基本属性外
            :: receipt_handle       下次删除或修改消息的临时句柄，next_visible_time之前有效
            :: next_visible_time    消息下次可消费时间

            @note: change_message_visibility 返回属性
            :: receipt_handle
            :: next_visible_time
        """
        self.message_body = "" if message_body is None else message_body
        self.delay_seconds = -1 if delay_seconds is None else delay_seconds
        self.priority = -1 if priority is None else priority

        self.message_id = ""
        self.message_body_md5 = ""

        self.dequeue_count = -1
        self.enqueue_time = -1
        self.first_dequeue_time = -1

        self.receipt_handle = ""
        self.next_visible_time = 1

    def set_delayseconds(self, delay_seconds):
        self.delay_seconds = delay_seconds

    def set_priority(self, priority):
        self.priority = priority