uamqp/__init__.py

#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------

# pylint: disable=no-member

import logging
import sys

from uamqp import c_uamqp  # pylint: disable=import-self

from uamqp.message import Message, BatchMessage
from uamqp.address import Source, Target

from uamqp.connection import Connection
from uamqp.session import Session
from uamqp.client import AMQPClient, SendClient, ReceiveClient
from uamqp.sender import MessageSender
from uamqp.receiver import MessageReceiver
from uamqp.constants import TransportType, MessageBodyType

try:
    from uamqp.async_ops import ConnectionAsync
    from uamqp.async_ops import SessionAsync
    from uamqp.async_ops import MessageSenderAsync
    from uamqp.async_ops import MessageReceiverAsync
    from uamqp.async_ops.client_async import (
        AMQPClientAsync,
        SendClientAsync,
        ReceiveClientAsync,
        AsyncMessageIter)
except (SyntaxError, ImportError):
    pass  # Async not supported.


__version__ = "1.6.11"


_logger = logging.getLogger(__name__)
_is_win = sys.platform.startswith('win')
c_uamqp.set_python_logger()


def send_message(target, data, auth=None, debug=False):
    """Send a single message to AMQP endpoint.

    :param target: The target AMQP endpoint.
    :type target: str, bytes or ~uamqp.address.Target
    :param data: The contents of the message to send.
    :type data: str, bytes or ~uamqp.message.Message
    :param auth: The authentication credentials for the endpoint.
     This should be one of the subclasses of uamqp.authentication.AMQPAuth. Currently
     this includes:
        - uamqp.authentication.SASLAnonymous
        - uamqp.authentication.SASLPlain
        - uamqp.authentication.SASTokenAuth
     If no authentication is supplied, SASLAnnoymous will be used by default.
    :type auth: ~uamqp.authentication.common.AMQPAuth
    :param debug: Whether to turn on network trace logs. If `True`, trace logs
     will be logged at INFO level. Default is `False`.
    :type debug: bool
    :return: A list of states for each message sent.
    :rtype: list[~uamqp.constants.MessageState]
    """
    message = data if isinstance(data, Message) else Message(body=data)
    with SendClient(target, auth=auth, debug=debug) as send_client:
        send_client.queue_message(message)  # pylint: disable=no-member
        return send_client.send_all_messages()  # pylint: disable=no-member


def receive_message(source, auth=None, timeout=0, debug=False):
    """Receive a single message from an AMQP endpoint.

    :param source: The AMQP source endpoint to receive from.
    :type source: str, bytes or ~uamqp.address.Source
    :param auth: The authentication credentials for the endpoint.
     This should be one of the subclasses of uamqp.authentication.AMQPAuth. Currently
     this includes:
        - uamqp.authentication.SASLAnonymous
        - uamqp.authentication.SASLPlain
        - uamqp.authentication.SASTokenAuth
     If no authentication is supplied, SASLAnnoymous will be used by default.
    :type auth: ~uamqp.authentication.common.AMQPAuth
    :param timeout: The timeout in milliseconds after which to return None if no messages
     are retrieved. If set to `0` (the default), the receiver will not timeout and
     will continue to wait for messages until interrupted.
    :param debug: Whether to turn on network trace logs. If `True`, trace logs
     will be logged at INFO level. Default is `False`.
    :type debug: bool
    :rtype: ~uamqp.message.Message or None
    """
    received = receive_messages(source, auth=auth, max_batch_size=1, timeout=timeout, debug=debug)
    if received:
        return received[0]
    return None


def receive_messages(source, auth=None, max_batch_size=None, timeout=0, debug=False, **kwargs):
    """Receive a batch of messages from an AMQP endpoint.

    :param source: The AMQP source endpoint to receive from.
    :type source: str, bytes or ~uamqp.address.Source
    :param auth: The authentication credentials for the endpoint.
     This should be one of the subclasses of ~uamqp.authentication.AMQPAuth. Currently
     this includes:
        - uamqp.authentication.SASLAnonymous
        - uamqp.authentication.SASLPlain
        - uamqp.authentication.SASTokenAuth
     If no authentication is supplied, SASLAnnoymous will be used by default.
    :type auth: ~uamqp.authentication.common.AMQPAuth
    :param max_batch_size: The maximum number of messages to return in a batch. If the
     receiver receives a smaller number than this, it will not wait to return them so
     the actual number returned can be anything up to this value. If the receiver reaches
     a timeout, an empty list will be returned.
    :param timeout: The timeout in milliseconds after which to return if no messages
     are retrieved. If set to `0` (the default), the receiver will not timeout and
     will continue to wait for messages until interrupted.
    :param debug: Whether to turn on network trace logs. If `True`, trace logs
     will be logged at INFO level. Default is `False`.
    :type debug: bool
    :rtype: list[~uamqp.message.Message]
    """
    if max_batch_size:
        kwargs['prefetch'] = max_batch_size
    with ReceiveClient(source, auth=auth, debug=debug, **kwargs) as receive_client:
        return receive_client.receive_message_batch(  # pylint: disable=no-member
            max_batch_size=max_batch_size or receive_client._prefetch, timeout=timeout)  # pylint: disable=protected-access, no-member


class _Platform(object):
    """Runs any platform preparatory steps for the AMQP C
    library. This is primarily used for OpenSSL setup.

    :ivar initialized: When the setup has completed.
    :vartype initialized: bool
    """

    initialized = False

    @classmethod
    def initialize(cls):
        """Initialize the TLS/SSL platform to prepare it for
        making AMQP requests. This only needs to happen once.
        """
        if cls.initialized:
            _logger.debug("Platform already initialized.")
        else:
            _logger.debug("Initializing platform.")
            c_uamqp.platform_init()
            cls.initialized = True

    @classmethod
    def deinitialize(cls):
        """Deinitialize the TLS/SSL platform to prepare it for
        making AMQP requests. This only needs to happen once.
        """
        if not cls.initialized:
            _logger.debug("Platform already deinitialized.")
        else:
            #cls.initialized = False
            _logger.debug("Deinitializing platform.")
            #c_uamqp.platform_deinit()


def get_platform_info():
    """Gets the current platform information.

    :rtype: str
    """
    return str(c_uamqp.get_info())