<commit_before>"""
AMQP 0-8 Client Library

"""
# Copyright (C) 2007 <NAME> <<EMAIL>>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA


import socket
from Queue import Queue
from struct import unpack
from util_0_8 import AMQPReader, AMQPWriter, ContentProperties

__all__ =  [
            'Connection',
            'Channel',      # here mainly so it shows in in pydoc
            'BasicContent',
            'AMQPException',
            'AMQPConnectionException',
            'AMQPChannelException',
           ]

AMQP_PORT = 5672
AMQP_PROTOCOL_HEADER = 'AMQP\x01\x01\x09\x01'

#
# Client property info that gets sent to the server on connection startup
#
LIBRARY_PROPERTIES = {
    'library': 'Python amqplib',
    'library_version': '0.1',
    }

DEBUG = False

class AMQPException(Exception):
    def __init__(self, reply_code, reply_text, method_sig):
        self.amqp_reply_code = reply_code
        self.amqp_reply_text = reply_text
        self.amqp_method_sig = method_sig
        self.args = (
            reply_code,
            reply_text,
            method_sig,
            _METHOD_NAME_MAP.get(method_sig, '')
            )


class AMQPConnectionException(AMQPException):
    pass


class AMQPChannelException(AMQPException):
    pass


class Connection(object):
    """
    The connection class provides methods for a client to establish a
    network connection to a server, and for both peers to operate the
    connection thereafter.

    GRAMMAR:

        connection          = open-connection *use-connection close-connection
        open-connection     = C:protocol-header
                              S:START C:START-OK
                              *challenge
                              S:TUNE C:TUNE-OK
                              C:OPEN S:OPEN-OK | S:REDIRECT
        challenge           = S:SECURE C:SECURE-OK
        use-connection      = *channel
        close-connection    = C:CLOSE S:CLOSE-OK
                            / S:CLOSE C:CLOSE-OK

    """
    def __init__(self, host, userid=None, password=<PASSWORD>, login_method='AMQPLAIN', login_response=None, virtual_host='/', locale='en_US', client_properties={}):
        """
        Create a connection to the specified host, which should be
        a 'host[:port]', such as 'localhost', or '1.2.3.4:5672'

        If a userid and password are specified, a login_response is built up for
        you.  Otherwise you have to roll your own.

        """
        self.channels = {}
        self.frame_queue = Queue()
        self.input = self.out = None

        if ':' in host:
            host, port = host.split(':', 1)
            port = int(port)
        else:
            port = AMQP_PORT

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((host, port))

        self.input = AMQPReader(sock.makefile('r'))
        self.out = sock.makefile('w')

        self.out.write(AMQP_PROTOCOL_HEADER)
        self.out.flush()

        self.wait(allowed_methods=[
                (10, 10), # start
                ])

        if (userid is not None) and (password is not None):
            login_response = AMQPWriter()
            login_response.write_table({'LOGIN': userid, 'PASSWORD': password})
            login_response = login_response.getvalue()[4:]    #Skip the length at the beginning

        d = {}
        d.update(LIBRARY_PROPERTIES)
        d.update(client_properties)
        self._x_start_ok(d, login_method, login_response, locale)

        self._wait_tune_ok = True
        while self._wait_tune_ok:
            self.wait(allowed_methods=[
                (10, 20), # secure
                (10, 30), # tune
                ])

        self._x_open(virtual_host)


    def __del__(self):
        if self.input is not None:
            self.close()

    def _do_close(self):
        self.input.close()
        self.out.close()
        self.input = self.out = None


    def _get_channel_queue(self, channel_id):
        if channel_id == 0:
            return self.frame_queue
        return self.channels[channel_id].frame_queue

    def _get_free_channel_id(self):
        for i in xrange(1, self.channel_max+1):
            if i not in self.channels:
                return i
        raise AMQPException('No free channel ids, current=%d, channel_max=%d' % (len(self.channels), self.channel_max))


    def _send_content(self, channel, class_id, weight, body_size, packed_properties, body):
        pkt = AMQPWriter()

        pkt.write_octet(2)
        pkt.write_short(channel)
        pkt.write_long(len(packed_properties)+12)

        pkt.write_short(class_id)
        pkt.write_short(weight)
        pkt.write_longlong(body_size)
        pkt.write(packed_properties)

        pkt.write_octet(0xce)
        pkt = pkt.getvalue()
        self.out.write(pkt)
        self.out.flush()

        while body:
            payload, body = body[:self.frame_max - 8], body[self.frame_max -8:]
            pkt = AMQPWriter()

            pkt.write_octet(3)
            pkt.write_short(channel)
            pkt.write_long(len(payload))

            pkt.write(payload)

            pkt.write_octet(0xce)
            pkt = pkt.getvalue()
            self.out.write(pkt)
            self.out.flush()


    def _send_method_frame(self, channel, method_sig, args=''):
        if isinstance(args, AMQPWriter):
            args = args.getvalue()

        pkt = AMQPWriter()

        pkt.write_octet(1)
        pkt.write_short(channel)
        pkt.write_long(len(args)+4)  # 4 = length of class_id and method_id in payload

        pkt.write_short(method_sig[0]) # class_id
        pkt.write_short(method_sig[1]) # method_id
        pkt.write(args)

        pkt.write_octet(0xce)
        pkt = pkt.getvalue()
#        _hexdump(pkt)
        self.out.write(pkt)
        self.out.flush()

        if DEBUG:
            print '> %s: %s' % (str(method_sig), _METHOD_NAME_MAP[method_sig])


    def _wait_frame(self):
        """
        Wait for a frame from the server

        """
        frame_type = self.input.read_octet()
        channel = self.input.read_short()
        size = self.input.read_long()
        payload = self.input.read(size)

        ch = self.input.read_octet()
        if ch != 0xce:
            raise Exception('Framing error, unexpected byte: %x' % ch)

        return frame_type, channel, payload


    def _wait_channel(self, channel_id):
        """
        Wait for a frame from the server sent to a particular
        channel.

        """
        queue = self._get_channel_queue(channel_id)
        if not queue.empty():
            return queue.get()

        while True:
            frame_type, frame_channel, payload = self._wait_frame()

            if frame_channel == channel_id:
                return frame_type, payload

            self._get_channel_queue(frame_channel).put((frame_type, payload))


    def _dispatch_method(self, channel, payload, allowed_methods):
        if len(payload) < 4:
            raise Exception('Method frame too short')

        method_sig = unpack('>HH', payload[:4])
        if DEBUG:
            print '< %s: %s' % (str(method_sig), _METHOD_NAME_MAP[method_sig])
        if allowed_methods and (method_sig not in allowed_methods):
            raise Exception('Received unexpected method: %s, was expecting one of: %s' % (method_sig, allowed_methods))

        args = AMQPReader(payload[4:])

        amqp_class, amqp_method = _METHOD_MAP.get(method_sig, (None, None))

        if amqp_class is Connection:
            return amqp_method(self, args)
        if amqp_class is Channel:
            ch = self.channels[channel]
            return amqp_method(ch, args)

        raise Exception('Unknown AMQP method (%d, %d)' % amqp_class, amqp_method)


    def channel(self, channel_id=None):
        """
        Fetch a Channel object identified by the numeric channel_id, or
        create that object of it doesn't already exist.

        """
        if channel_id in self.channels:
            return self.channels[channel_id]

        return Channel(self, channel_id)


    def wait(self, allowed_methods=None):
        """
        Not sure yet if this should even be a public method.

        """
        frame_type, payload = self._wait_channel(0)

        if frame_type == 1:
            return self._dispatch_method(0, payload, allowed_methods)


    #################

    def close(self, reply_code=0, reply_text='', method_sig=(0, 0)):
        """
        This method indicates that the sender wants to close the connection.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        args = AMQPWriter()
        args.write_short(reply_code)
        args.write_shortstr(reply_text)
        args.write_short(method_sig[0]) # class_id
        args.write_short(method_sig[1]) # method_id
        self._send_method_frame(0, (10, 60), args)
        return self.wait(allowed_methods=[
                          (10, 61),    # Connection.close_ok
                        ])


    def _close(self, args):
        """
        This method indicates that the sender wants to close the connection.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        reply_code = args.read_short()
        reply_text = args.read_shortstr()
        class_id = args.read_short()
        method_id = args.read_short()

        self._x_close_ok()

        raise AMQPConnectionException(reply_code, reply_text, (class_id, method_id))


    def _x_close_ok(self):
        """
        This method confirms a Connection.Close method and tells the
        recipient that it is safe to release resources for the connection
        and close the socket.

        """
        self._send_method_frame(0, (10, 61))
        self._do_close()


    def _close_ok(self, args):
        """
        This method confirms a Connection.Close method and tells the
        recipient that it is safe to release resources for the connection
        and close the socket.

        """
        self._do_close()


    def _x_open(self, virtual_host, capabilities='', insist=False):
        """
        This method opens a connection to a virtual host, which is a
        collection of resources, and acts to separate multiple application
        domains within a server.

        """
        args = AMQPWriter()
        args.write_shortstr(virtual_host)
        args.write_shortstr(capabilities)
        args.write_bit(insist)
        self._send_method_frame(0, (10, 40), args)
        return self.wait(allowed_methods=[
                          (10, 41),    # Connection.open_ok
                          (10, 50),    # Connection.redirect
                        ])


    def _open_ok(self, args):
        """
        This method signals to the client that the connection is ready for
        use.

        """
        self.known_hosts = args.read_shortstr()
        if DEBUG:
            print 'Open OK! known_hosts [%s]' % self.known_hosts


    def _redirect(self, args):
        """
        This method redirects the client to another server, based on the
        requested virtual host and/or capabilities.

        """
        host = args.read_shortstr()
        known_hosts = args.read_shortstr()


    def _secure(self, args):
        """
        The SASL protocol works by exchanging challenges and responses until
        both peers have received sufficient information to authenticate each
        other.  This method challenges the client to provide more information.

        """
        challenge = args.read_longstr()


    def _x_secure_ok(self, response):
        """
        This method attempts to authenticate, passing a block of SASL data
        for the security mechanism at the server side.

        """
        args = AMQPWriter()
        args.write_longstr(response)
        self._send_method_frame(0, (10, 21), args)


    def _start(self, args):
        """
        This method starts the connection negotiation process by telling
        the client the protocol version that the server proposes, along
        with a list of security mechanisms which the client can use for
        authentication.

        """
        self.version_major = args.read_octet()
        self.version_minor = args.read_octet()
        self.server_properties = args.read_table()
        self.mechanisms = args.read_longstr().split(' ')
        self.locales = args.read_longstr().split(' ')

        if DEBUG:
            print 'Start from server, version: %d.%d, properties: %s, mechanisms: %s, locales: %s' % (self.version_major, self.version_minor, str(self.server_properties), self.mechanisms, self.locales)


    def _x_start_ok(self, client_properties, mechanism, response, locale):
        """
        This method selects a SASL security mechanism. ASL uses SASL
        (RFC2222) to negotiate authentication and encryption.

        """
        args = AMQPWriter()
        args.write_table(client_properties)
        args.write_shortstr(mechanism)
        args.write_longstr(response)
        args.write_shortstr(locale)
        self._send_method_frame(0, (10, 11), args)


    def _tune(self, args):
        """
        This method proposes a set of connection configuration values
        to the client.  The client can accept and/or adjust these.

        """
        self.channel_max = args.read_short() or 65535
        self.frame_max = args.read_long() or 131072
        self.heartbeat = args.read_short()

        self._x_tune_ok(self.channel_max, self.frame_max, 0)


    def _x_tune_ok(self, channel_max, frame_max, heartbeat):
        """
        This method sends the client's connection tuning parameters to the
        server. Certain fields are negotiated, others provide capability
        information.

        """
        args = AMQPWriter()
        args.write_short(channel_max)
        args.write_long(frame_max)
        args.write_short(heartbeat)
        self._send_method_frame(0, (10, 31), args)
        self._wait_tune_ok = False


class Channel(object):
    """
    The channel class provides methods for a client to establish a virtual
    connection - a channel - to a server and for both peers to operate the
    virtual connection thereafter.

    GRAMMAR:

        channel             = open-channel *use-channel close-channel
        open-channel        = C:OPEN S:OPEN-OK
        use-channel         = C:FLOW S:FLOW-OK
                            / S:FLOW C:FLOW-OK
                            / S:ALERT
                            / functional-class
        close-channel       = C:CLOSE S:CLOSE-OK
                            / S:CLOSE C:CLOSE-OK

    """
    def __init__(self, connection, channel_id=None):
        """
        Create a channel bound to a connection and using the specified
        numeric channel_id, and open on the server.

        """
        self.connection = connection
        if channel_id is None:
            channel_id = connection._get_free_channel_id()
        if DEBUG:
            print 'using channel_id', channel_id
        self.channel_id = channel_id
        self.default_ticket = 0
        self.is_open = False
        self.active = True # Flow control
        connection.channels[channel_id] = self
        self.frame_queue = Queue()
        self.alerts = Queue()
        self.callbacks = {}

        self._x_open()


    def __del__(self):
        if self.connection:
            self.close(msg='destroying channel')


    def _do_close(self):
        """
        Tear down this object, after we've agreed to close with the server.

        """
        if DEBUG:
            print 'Closed channel #%d' % self.channel_id
        self.is_open = False
        del self.connection.channels[self.channel_id]
        self.channel_id = self.connection = None


    def wait(self, allowed_methods=None):
        frame_type, payload = self.connection._wait_channel(self.channel_id)

        if frame_type == 1:
            return self.connection._dispatch_method(self.channel_id, payload, allowed_methods)
        if frame_type == 2:
            class_id, weight, body_size = unpack('>HHQ', payload[:12])
            content_properties = BASIC_CONTENT_PROPERTIES.parse(payload[12:])

            body_parts = []
            body_received = 0
            while body_received < body_size:
                frame_type, payload = self.connection._wait_channel(self.channel_id)
                if frame_type == 3:
                    body_parts.append(payload)
                    body_received += len(payload)

            return BasicContent(''.join(body_parts), **content_properties)


    def _send_method_frame(self, method_sig, args=''):
        self.connection._send_method_frame(self.channel_id, method_sig, args)

    #################

    def _alert(self, args):
        """
        This method allows the server to send a non-fatal warning to the
        client.  This is used for methods that are normally asynchronous
        and thus do not have confirmations, and for which the server may
        detect errors that need to be reported.  Fatal errors are handled
        as channel or connection exceptions; non-fatal errors are sent
        through this method.

        """
        reply_code = args.read_short()
        reply_text = args.read_shortstr()
        details = args.read_table()

        self.alerts.put((reply_code, reply_text, details))


    def close(self, reply_code=0, reply_text='', method_sig=(0, 0)):
        """
        This method indicates that the sender wants to close the channel.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        args = AMQPWriter()
        args.write_short(reply_code)
        args.write_shortstr(reply_text)
        args.write_short(method_sig[0]) # class_id
        args.write_short(method_sig[1]) # method_id
        self._send_method_frame((20, 40), args)
        return self.wait(allowed_methods=[
                          (20, 41),    # Channel.close_ok
                        ])


    def _close(self, args):
        """
        This method indicates that the sender wants to close the channel.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        reply_code = args.read_short()
        reply_text = args.read_shortstr()
        class_id = args.read_short()
        method_id = args.read_short()

#        self.close_ok()


#    def close_ok(self):
#        """
#        This method confirms a Channel.Close method and tells the recipient
#        that it is safe to release resources for the channel and close the
#        socket.
#
#        """
        self._send_method_frame((20, 41))
        self._do_close()

        raise AMQPChannelException(reply_code, reply_text, (class_id, method_id))


    def _close_ok(self, args):
        """
        This method confirms a Channel.Close method and tells the recipient
        that it is safe to release resources for the channel and close the
        socket.

        """
        self._do_close()


    def flow(self, active):
        """
        This method asks the peer to pause or restart the flow of content
        data. This is a simple flow-control mechanism that a peer can use
        to avoid oveflowing its queues or otherwise finding itself receiving
        more messages than it can process.  Note that this method is not
        intended for window control.  The peer that receives a request to
        stop sending content should finish sending the current content, if
        any, and then wait until it receives a Flow restart method.

        """
        args = AMQPWriter()
        args.write_bit(active)
        self._send_method_frame((20, 20), args)
        return self.wait(allowed_methods=[
                          (20, 21),    # Channel.flow_ok
                        ])


    def _flow(self, args):
        """
        This method asks the peer to pause or restart the flow of content
        data. This is a simple flow-control mechanism that a peer can use
        to avoid oveflowing its queues or otherwise finding itself receiving
        more messages than it can process.  Note that this method is not
        intended for window control.  The peer that receives a request to
        stop sending content should finish sending the current content, if
        any, and then wait until it receives a Flow restart method.

        """
        self.active = args.read_bit()

        self._x_flow_ok(self.active)


    def _x_flow_ok(self, active):
        """
        Confirms to the peer that a flow command was received and processed.

        """
        args = AMQPWriter()
        args.write_bit(active)
        self._send_method_frame((20, 21), args)


    def _flow_ok(self, args):
        """
        Confirms to the peer that a flow command was received and processed.

        """
        return args.read_bit()


    def _x_open(self, out_of_band=''):
        """
        This method opens a virtual connection (a channel).

        """
        if self.is_open:
            return

        args = AMQPWriter()
        args.write_shortstr(out_of_band)
        self._send_method_frame((20, 10), args)
        return self.wait(allowed_methods=[
                          (20, 11),    # Channel.open_ok
                        ])


    def _open_ok(self, args):
        """
        This method signals to the client that the channel is ready for use.

        """
        self.is_open = True
        if DEBUG:
            print 'Channel open'


    #############
    #
    #  Access
    #
    #
    # The protocol control access to server resources using access tickets.
    # A client must explicitly request access tickets before doing work.
    # An access ticket grants a client the right to use a specific set of
    # resources - called a "realm" - in specific ways.
    #
    # GRAMMAR:
    #
    #     access              = C:REQUEST S:REQUEST-OK
    #
    #

    def access_request(self, realm, exclusive=False, passive=False, active=False, write=False, read=False):
        """
        This method requests an access ticket for an access realm.
        The server responds by granting the access ticket.  If the
        client does not have access rights to the requested realm
        this causes a connection exception.  Access tickets are a
        per-channel resource.

        The most recently requested ticket is used as the channel's
        default ticket for any method that requires a ticket.

        """
        args = AMQPWriter()
        args.write_shortstr(realm)
        args.write_bit(exclusive)
        args.write_bit(passive)
        args.write_bit(active)
        args.write_bit(write)
        args.write_bit(read)
        self._send_method_frame((30, 10), args)
        return self.wait(allowed_methods=[
                          (30, 11),    # Channel.access_request_ok
                        ])


    def _access_request_ok(self, args):
        """
        This method provides the client with an access ticket. The access
        ticket is valid within the current channel and for the lifespan of
        the channel.

        """
        self.default_ticket = args.read_short()
        return self.default_ticket


    #############
    #
    #  Exchange
    #
    #
    # Exchanges match and distribute messages across queues.  Exchanges can be
    # configured in the server or created at runtime.
    #
    # GRAMMAR:
    #
    #     exchange            = C:DECLARE  S:DECLARE-OK
    #                         / C:DELETE   S:DELETE-OK
    #
    #

    def exchange_declare(self, exchange, type, passive=False, durable=False, auto_delete=False, internal=False, nowait=False, arguments={}, ticket=None):
        """
        This method creates an exchange if it does not already exist, and if the
        exchange exists, verifies that it is of the correct and expected class.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(exchange)
        args.write_shortstr(type)
        args.write_bit(passive)
        args.write_bit(durable)
        args.write_bit(auto_delete)
        args.write_bit(internal)
        args.write_bit(nowait)
        args.write_table(arguments)
        self._send_method_frame((40, 10), args)
        return self.wait(allowed_methods=[
                          (40, 11),    # Channel.exchange_declare_ok
                        ])


    def _exchange_declare_ok(self, args):
        """
        This method confirms a Declare method and confirms the name of the
        exchange, essential for automatically-named exchanges.

        """
        pass


    def exchange_delete(self, exchange, if_unused=False, nowait=False, ticket=None):
        """
        This method deletes an exchange.  When an exchange is deleted all queue
        bindings on the exchange are cancelled.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(exchange)
        args.write_bit(if_unused)
        args.write_bit(nowait)
        self._send_method_frame((40, 20), args)
        return self.wait(allowed_methods=[
                          (40, 21),    # Channel.exchange_delete_ok
                        ])


    def _exchange_delete_ok(self, args):
        """
        This method confirms the deletion of an exchange.

        """
        pass


    #############
    #
    #  Queue
    #
    #
    # Queues store and forward messages.  Queues can be configured in the server
    # or created at runtime.  Queues must be attached to at least one exchange
    # in order to receive messages from publishers.
    #
    # GRAMMAR:
    #
    #     queue               = C:DECLARE  S:DECLARE-OK
    #                         / C:BIND     S:BIND-OK
    #                         / C:PURGE    S:PURGE-OK
    #                         / C:DELETE   S:DELETE-OK
    #
    #

    def queue_bind(self, queue, exchange, routing_key='', nowait=False, arguments={}, ticket=None):
        """
        This method binds a queue to an exchange.  Until a queue is
        bound it will not receive any messages.  In a classic messaging
        model, store-and-forward queues are bound to a dest exchange
        and subscription queues are bound to a dest_wild exchange.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_shortstr(exchange)
        args.write_shortstr(routing_key)
        args.write_bit(nowait)
        args.write_table(arguments)
        self._send_method_frame((50, 20), args)
        return self.wait(allowed_methods=[
                          (50, 21),    # Channel.queue_bind_ok
                        ])


    def _queue_bind_ok(self, args):
        """
        This method confirms that the bind was successful.

        """
        pass


    def queue_declare(self, queue='', passive=False, durable=False, exclusive=False, auto_delete=False, nowait=False, arguments={}, ticket=None):
        """
        This method creates or checks a queue.  When creating a new queue
        the client can specify various properties that control the durability
        of the queue and its contents, and the level of sharing for the queue.

        Returns a tuple containing 3 items:
            the name of the queue (essential for automatically-named queues)
            message count
            consumer count

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_bit(passive)
        args.write_bit(durable)
        args.write_bit(exclusive)
        args.write_bit(auto_delete)
        args.write_bit(nowait)
        args.write_table(arguments)
        self._send_method_frame((50, 10), args)
        return self.wait(allowed_methods=[
                          (50, 11),    # Channel.queue_declare_ok
                        ])


    def _queue_declare_ok(self, args):
        """
        This method confirms a Declare method and confirms the name of the
        queue, essential for automatically-named queues.

        """
        queue = args.read_shortstr()
        message_count = args.read_long()
        consumer_count = args.read_long()

        return queue, message_count, consumer_count


    def queue_delete(self, queue, if_unused=False, if_empty=False, nowait=False, ticket=None):
        """
        This method deletes a queue.  When a queue is deleted any pending
        messages are sent to a dead-letter queue if this is defined in the
        server configuration, and all consumers on the queue are cancelled.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_bit(if_unused)
        args.write_bit(if_empty)
        args.write_bit(nowait)
        self._send_method_frame((50, 40), args)
        return self.wait(allowed_methods=[
                          (50, 41),    # Channel.queue_delete_ok
                        ])


    def _queue_delete_ok(self, args):
        """
        This method confirms the deletion of a queue.

        """
        return args.read_long()


    def queue_purge(self, queue, nowait=False, ticket=None):
        """
        This method removes all messages from a queue.  It does not cancel
        consumers.  Purged messages are deleted without any formal "undo"
        mechanism.

        if nowait is False, returns a message_count

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_bit(nowait)
        self._send_method_frame((50, 30), args)
        return self.wait(allowed_methods=[
                          (50, 31),    # Channel.queue_purge_ok
                        ])


    def _queue_purge_ok(self, args):
        """
        This method confirms the purge of a queue.

        """
        return args.read_long()


    #############
    #
    #  Basic
    #
    #
    # The Basic class provides methods that support an industry-standard
    # messaging model.
    #
    # GRAMMAR:
    #
    #     basic               = C:QOS S:QOS-OK
    #                         / C:CONSUME S:CONSUME-OK
    #                         / C:CANCEL S:CANCEL-OK
    #                         / C:PUBLISH content
    #                         / S:RETURN content
    #                         / S:DELIVER content
    #                         / C:GET ( S:GET-OK content / S:GET-EMPTY )
    #                         / C:ACK
    #                         / C:REJECT<commit_msg>Put back in the processing of 'nowait' args, that got lost in r35<commit_after>"""
AMQP 0-8 Client Library

"""
# Copyright (C) 2007 <NAME> <<EMAIL>>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA


import socket
from Queue import Queue
from struct import unpack
from util_0_8 import AMQPReader, AMQPWriter, ContentProperties

__all__ =  [
            'Connection',
            'Channel',      # here mainly so it shows in in pydoc
            'BasicContent',
            'AMQPException',
            'AMQPConnectionException',
            'AMQPChannelException',
           ]

AMQP_PORT = 5672
AMQP_PROTOCOL_HEADER = 'AMQP\x01\x01\x09\x01'

#
# Client property info that gets sent to the server on connection startup
#
LIBRARY_PROPERTIES = {
    'library': 'Python amqplib',
    'library_version': '0.1',
    }

DEBUG = False

class AMQPException(Exception):
    def __init__(self, reply_code, reply_text, method_sig):
        self.amqp_reply_code = reply_code
        self.amqp_reply_text = reply_text
        self.amqp_method_sig = method_sig
        self.args = (
            reply_code,
            reply_text,
            method_sig,
            _METHOD_NAME_MAP.get(method_sig, '')
            )


class AMQPConnectionException(AMQPException):
    pass


class AMQPChannelException(AMQPException):
    pass


class Connection(object):
    """
    The connection class provides methods for a client to establish a
    network connection to a server, and for both peers to operate the
    connection thereafter.

    GRAMMAR:

        connection          = open-connection *use-connection close-connection
        open-connection     = C:protocol-header
                              S:START C:START-OK
                              *challenge
                              S:TUNE C:TUNE-OK
                              C:OPEN S:OPEN-OK | S:REDIRECT
        challenge           = S:SECURE C:SECURE-OK
        use-connection      = *channel
        close-connection    = C:CLOSE S:CLOSE-OK
                            / S:CLOSE C:CLOSE-OK

    """
    def __init__(self, host, userid=None, password=<PASSWORD>, login_method='AMQPLAIN', login_response=None, virtual_host='/', locale='en_US', client_properties={}):
        """
        Create a connection to the specified host, which should be
        a 'host[:port]', such as 'localhost', or '1.2.3.4:5672'

        If a userid and password are specified, a login_response is built up for
        you.  Otherwise you have to roll your own.

        """
        self.channels = {}
        self.frame_queue = Queue()
        self.input = self.out = None

        if ':' in host:
            host, port = host.split(':', 1)
            port = int(port)
        else:
            port = AMQP_PORT

        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((host, port))

        self.input = AMQPReader(sock.makefile('r'))
        self.out = sock.makefile('w')

        self.out.write(AMQP_PROTOCOL_HEADER)
        self.out.flush()

        self.wait(allowed_methods=[
                (10, 10), # start
                ])

        if (userid is not None) and (password is not None):
            login_response = AMQPWriter()
            login_response.write_table({'LOGIN': userid, 'PASSWORD': password})
            login_response = login_response.getvalue()[4:]    #Skip the length at the beginning

        d = {}
        d.update(LIBRARY_PROPERTIES)
        d.update(client_properties)
        self._x_start_ok(d, login_method, login_response, locale)

        self._wait_tune_ok = True
        while self._wait_tune_ok:
            self.wait(allowed_methods=[
                (10, 20), # secure
                (10, 30), # tune
                ])

        self._x_open(virtual_host)


    def __del__(self):
        if self.input is not None:
            self.close()

    def _do_close(self):
        self.input.close()
        self.out.close()
        self.input = self.out = None


    def _get_channel_queue(self, channel_id):
        if channel_id == 0:
            return self.frame_queue
        return self.channels[channel_id].frame_queue

    def _get_free_channel_id(self):
        for i in xrange(1, self.channel_max+1):
            if i not in self.channels:
                return i
        raise AMQPException('No free channel ids, current=%d, channel_max=%d' % (len(self.channels), self.channel_max))


    def _send_content(self, channel, class_id, weight, body_size, packed_properties, body):
        pkt = AMQPWriter()

        pkt.write_octet(2)
        pkt.write_short(channel)
        pkt.write_long(len(packed_properties)+12)

        pkt.write_short(class_id)
        pkt.write_short(weight)
        pkt.write_longlong(body_size)
        pkt.write(packed_properties)

        pkt.write_octet(0xce)
        pkt = pkt.getvalue()
        self.out.write(pkt)
        self.out.flush()

        while body:
            payload, body = body[:self.frame_max - 8], body[self.frame_max -8:]
            pkt = AMQPWriter()

            pkt.write_octet(3)
            pkt.write_short(channel)
            pkt.write_long(len(payload))

            pkt.write(payload)

            pkt.write_octet(0xce)
            pkt = pkt.getvalue()
            self.out.write(pkt)
            self.out.flush()


    def _send_method_frame(self, channel, method_sig, args=''):
        if isinstance(args, AMQPWriter):
            args = args.getvalue()

        pkt = AMQPWriter()

        pkt.write_octet(1)
        pkt.write_short(channel)
        pkt.write_long(len(args)+4)  # 4 = length of class_id and method_id in payload

        pkt.write_short(method_sig[0]) # class_id
        pkt.write_short(method_sig[1]) # method_id
        pkt.write(args)

        pkt.write_octet(0xce)
        pkt = pkt.getvalue()
#        _hexdump(pkt)
        self.out.write(pkt)
        self.out.flush()

        if DEBUG:
            print '> %s: %s' % (str(method_sig), _METHOD_NAME_MAP[method_sig])


    def _wait_frame(self):
        """
        Wait for a frame from the server

        """
        frame_type = self.input.read_octet()
        channel = self.input.read_short()
        size = self.input.read_long()
        payload = self.input.read(size)

        ch = self.input.read_octet()
        if ch != 0xce:
            raise Exception('Framing error, unexpected byte: %x' % ch)

        return frame_type, channel, payload


    def _wait_channel(self, channel_id):
        """
        Wait for a frame from the server sent to a particular
        channel.

        """
        queue = self._get_channel_queue(channel_id)
        if not queue.empty():
            return queue.get()

        while True:
            frame_type, frame_channel, payload = self._wait_frame()

            if frame_channel == channel_id:
                return frame_type, payload

            self._get_channel_queue(frame_channel).put((frame_type, payload))


    def _dispatch_method(self, channel, payload, allowed_methods):
        if len(payload) < 4:
            raise Exception('Method frame too short')

        method_sig = unpack('>HH', payload[:4])
        if DEBUG:
            print '< %s: %s' % (str(method_sig), _METHOD_NAME_MAP[method_sig])
        if allowed_methods and (method_sig not in allowed_methods):
            raise Exception('Received unexpected method: %s, was expecting one of: %s' % (method_sig, allowed_methods))

        args = AMQPReader(payload[4:])

        amqp_class, amqp_method = _METHOD_MAP.get(method_sig, (None, None))

        if amqp_class is Connection:
            return amqp_method(self, args)
        if amqp_class is Channel:
            ch = self.channels[channel]
            return amqp_method(ch, args)

        raise Exception('Unknown AMQP method (%d, %d)' % amqp_class, amqp_method)


    def channel(self, channel_id=None):
        """
        Fetch a Channel object identified by the numeric channel_id, or
        create that object of it doesn't already exist.

        """
        if channel_id in self.channels:
            return self.channels[channel_id]

        return Channel(self, channel_id)


    def wait(self, allowed_methods=None):
        """
        Not sure yet if this should even be a public method.

        """
        frame_type, payload = self._wait_channel(0)

        if frame_type == 1:
            return self._dispatch_method(0, payload, allowed_methods)


    #################

    def close(self, reply_code=0, reply_text='', method_sig=(0, 0)):
        """
        This method indicates that the sender wants to close the connection.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        args = AMQPWriter()
        args.write_short(reply_code)
        args.write_shortstr(reply_text)
        args.write_short(method_sig[0]) # class_id
        args.write_short(method_sig[1]) # method_id
        self._send_method_frame(0, (10, 60), args)
        return self.wait(allowed_methods=[
                          (10, 61),    # Connection.close_ok
                        ])


    def _close(self, args):
        """
        This method indicates that the sender wants to close the connection.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        reply_code = args.read_short()
        reply_text = args.read_shortstr()
        class_id = args.read_short()
        method_id = args.read_short()

        self._x_close_ok()

        raise AMQPConnectionException(reply_code, reply_text, (class_id, method_id))


    def _x_close_ok(self):
        """
        This method confirms a Connection.Close method and tells the
        recipient that it is safe to release resources for the connection
        and close the socket.

        """
        self._send_method_frame(0, (10, 61))
        self._do_close()


    def _close_ok(self, args):
        """
        This method confirms a Connection.Close method and tells the
        recipient that it is safe to release resources for the connection
        and close the socket.

        """
        self._do_close()


    def _x_open(self, virtual_host, capabilities='', insist=False):
        """
        This method opens a connection to a virtual host, which is a
        collection of resources, and acts to separate multiple application
        domains within a server.

        """
        args = AMQPWriter()
        args.write_shortstr(virtual_host)
        args.write_shortstr(capabilities)
        args.write_bit(insist)
        self._send_method_frame(0, (10, 40), args)
        return self.wait(allowed_methods=[
                          (10, 41),    # Connection.open_ok
                          (10, 50),    # Connection.redirect
                        ])


    def _open_ok(self, args):
        """
        This method signals to the client that the connection is ready for
        use.

        """
        self.known_hosts = args.read_shortstr()
        if DEBUG:
            print 'Open OK! known_hosts [%s]' % self.known_hosts


    def _redirect(self, args):
        """
        This method redirects the client to another server, based on the
        requested virtual host and/or capabilities.

        """
        host = args.read_shortstr()
        known_hosts = args.read_shortstr()


    def _secure(self, args):
        """
        The SASL protocol works by exchanging challenges and responses until
        both peers have received sufficient information to authenticate each
        other.  This method challenges the client to provide more information.

        """
        challenge = args.read_longstr()


    def _x_secure_ok(self, response):
        """
        This method attempts to authenticate, passing a block of SASL data
        for the security mechanism at the server side.

        """
        args = AMQPWriter()
        args.write_longstr(response)
        self._send_method_frame(0, (10, 21), args)


    def _start(self, args):
        """
        This method starts the connection negotiation process by telling
        the client the protocol version that the server proposes, along
        with a list of security mechanisms which the client can use for
        authentication.

        """
        self.version_major = args.read_octet()
        self.version_minor = args.read_octet()
        self.server_properties = args.read_table()
        self.mechanisms = args.read_longstr().split(' ')
        self.locales = args.read_longstr().split(' ')

        if DEBUG:
            print 'Start from server, version: %d.%d, properties: %s, mechanisms: %s, locales: %s' % (self.version_major, self.version_minor, str(self.server_properties), self.mechanisms, self.locales)


    def _x_start_ok(self, client_properties, mechanism, response, locale):
        """
        This method selects a SASL security mechanism. ASL uses SASL
        (RFC2222) to negotiate authentication and encryption.

        """
        args = AMQPWriter()
        args.write_table(client_properties)
        args.write_shortstr(mechanism)
        args.write_longstr(response)
        args.write_shortstr(locale)
        self._send_method_frame(0, (10, 11), args)


    def _tune(self, args):
        """
        This method proposes a set of connection configuration values
        to the client.  The client can accept and/or adjust these.

        """
        self.channel_max = args.read_short() or 65535
        self.frame_max = args.read_long() or 131072
        self.heartbeat = args.read_short()

        self._x_tune_ok(self.channel_max, self.frame_max, 0)


    def _x_tune_ok(self, channel_max, frame_max, heartbeat):
        """
        This method sends the client's connection tuning parameters to the
        server. Certain fields are negotiated, others provide capability
        information.

        """
        args = AMQPWriter()
        args.write_short(channel_max)
        args.write_long(frame_max)
        args.write_short(heartbeat)
        self._send_method_frame(0, (10, 31), args)
        self._wait_tune_ok = False


class Channel(object):
    """
    The channel class provides methods for a client to establish a virtual
    connection - a channel - to a server and for both peers to operate the
    virtual connection thereafter.

    GRAMMAR:

        channel             = open-channel *use-channel close-channel
        open-channel        = C:OPEN S:OPEN-OK
        use-channel         = C:FLOW S:FLOW-OK
                            / S:FLOW C:FLOW-OK
                            / S:ALERT
                            / functional-class
        close-channel       = C:CLOSE S:CLOSE-OK
                            / S:CLOSE C:CLOSE-OK

    """
    def __init__(self, connection, channel_id=None):
        """
        Create a channel bound to a connection and using the specified
        numeric channel_id, and open on the server.

        """
        self.connection = connection
        if channel_id is None:
            channel_id = connection._get_free_channel_id()
        if DEBUG:
            print 'using channel_id', channel_id
        self.channel_id = channel_id
        self.default_ticket = 0
        self.is_open = False
        self.active = True # Flow control
        connection.channels[channel_id] = self
        self.frame_queue = Queue()
        self.alerts = Queue()
        self.callbacks = {}

        self._x_open()


    def __del__(self):
        if self.connection:
            self.close(msg='destroying channel')


    def _do_close(self):
        """
        Tear down this object, after we've agreed to close with the server.

        """
        if DEBUG:
            print 'Closed channel #%d' % self.channel_id
        self.is_open = False
        del self.connection.channels[self.channel_id]
        self.channel_id = self.connection = None


    def wait(self, allowed_methods=None):
        frame_type, payload = self.connection._wait_channel(self.channel_id)

        if frame_type == 1:
            return self.connection._dispatch_method(self.channel_id, payload, allowed_methods)
        if frame_type == 2:
            class_id, weight, body_size = unpack('>HHQ', payload[:12])
            content_properties = BASIC_CONTENT_PROPERTIES.parse(payload[12:])

            body_parts = []
            body_received = 0
            while body_received < body_size:
                frame_type, payload = self.connection._wait_channel(self.channel_id)
                if frame_type == 3:
                    body_parts.append(payload)
                    body_received += len(payload)

            return BasicContent(''.join(body_parts), **content_properties)


    def _send_method_frame(self, method_sig, args=''):
        self.connection._send_method_frame(self.channel_id, method_sig, args)

    #################

    def _alert(self, args):
        """
        This method allows the server to send a non-fatal warning to the
        client.  This is used for methods that are normally asynchronous
        and thus do not have confirmations, and for which the server may
        detect errors that need to be reported.  Fatal errors are handled
        as channel or connection exceptions; non-fatal errors are sent
        through this method.

        """
        reply_code = args.read_short()
        reply_text = args.read_shortstr()
        details = args.read_table()

        self.alerts.put((reply_code, reply_text, details))


    def close(self, reply_code=0, reply_text='', method_sig=(0, 0)):
        """
        This method indicates that the sender wants to close the channel.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        args = AMQPWriter()
        args.write_short(reply_code)
        args.write_shortstr(reply_text)
        args.write_short(method_sig[0]) # class_id
        args.write_short(method_sig[1]) # method_id
        self._send_method_frame((20, 40), args)
        return self.wait(allowed_methods=[
                          (20, 41),    # Channel.close_ok
                        ])


    def _close(self, args):
        """
        This method indicates that the sender wants to close the channel.
        This may be due to internal conditions (e.g. a forced shut-down) or
        due to an error handling a specific method, i.e. an exception.  When
        a close is due to an exception, the sender provides the class and
        method id of the method which caused the exception.

        """
        reply_code = args.read_short()
        reply_text = args.read_shortstr()
        class_id = args.read_short()
        method_id = args.read_short()

#        self.close_ok()


#    def close_ok(self):
#        """
#        This method confirms a Channel.Close method and tells the recipient
#        that it is safe to release resources for the channel and close the
#        socket.
#
#        """
        self._send_method_frame((20, 41))
        self._do_close()

        raise AMQPChannelException(reply_code, reply_text, (class_id, method_id))


    def _close_ok(self, args):
        """
        This method confirms a Channel.Close method and tells the recipient
        that it is safe to release resources for the channel and close the
        socket.

        """
        self._do_close()


    def flow(self, active):
        """
        This method asks the peer to pause or restart the flow of content
        data. This is a simple flow-control mechanism that a peer can use
        to avoid oveflowing its queues or otherwise finding itself receiving
        more messages than it can process.  Note that this method is not
        intended for window control.  The peer that receives a request to
        stop sending content should finish sending the current content, if
        any, and then wait until it receives a Flow restart method.

        """
        args = AMQPWriter()
        args.write_bit(active)
        self._send_method_frame((20, 20), args)
        return self.wait(allowed_methods=[
                          (20, 21),    # Channel.flow_ok
                        ])


    def _flow(self, args):
        """
        This method asks the peer to pause or restart the flow of content
        data. This is a simple flow-control mechanism that a peer can use
        to avoid oveflowing its queues or otherwise finding itself receiving
        more messages than it can process.  Note that this method is not
        intended for window control.  The peer that receives a request to
        stop sending content should finish sending the current content, if
        any, and then wait until it receives a Flow restart method.

        """
        self.active = args.read_bit()

        self._x_flow_ok(self.active)


    def _x_flow_ok(self, active):
        """
        Confirms to the peer that a flow command was received and processed.

        """
        args = AMQPWriter()
        args.write_bit(active)
        self._send_method_frame((20, 21), args)


    def _flow_ok(self, args):
        """
        Confirms to the peer that a flow command was received and processed.

        """
        return args.read_bit()


    def _x_open(self, out_of_band=''):
        """
        This method opens a virtual connection (a channel).

        """
        if self.is_open:
            return

        args = AMQPWriter()
        args.write_shortstr(out_of_band)
        self._send_method_frame((20, 10), args)
        return self.wait(allowed_methods=[
                          (20, 11),    # Channel.open_ok
                        ])


    def _open_ok(self, args):
        """
        This method signals to the client that the channel is ready for use.

        """
        self.is_open = True
        if DEBUG:
            print 'Channel open'


    #############
    #
    #  Access
    #
    #
    # The protocol control access to server resources using access tickets.
    # A client must explicitly request access tickets before doing work.
    # An access ticket grants a client the right to use a specific set of
    # resources - called a "realm" - in specific ways.
    #
    # GRAMMAR:
    #
    #     access              = C:REQUEST S:REQUEST-OK
    #
    #

    def access_request(self, realm, exclusive=False, passive=False, active=False, write=False, read=False):
        """
        This method requests an access ticket for an access realm.
        The server responds by granting the access ticket.  If the
        client does not have access rights to the requested realm
        this causes a connection exception.  Access tickets are a
        per-channel resource.

        The most recently requested ticket is used as the channel's
        default ticket for any method that requires a ticket.

        """
        args = AMQPWriter()
        args.write_shortstr(realm)
        args.write_bit(exclusive)
        args.write_bit(passive)
        args.write_bit(active)
        args.write_bit(write)
        args.write_bit(read)
        self._send_method_frame((30, 10), args)
        return self.wait(allowed_methods=[
                          (30, 11),    # Channel.access_request_ok
                        ])


    def _access_request_ok(self, args):
        """
        This method provides the client with an access ticket. The access
        ticket is valid within the current channel and for the lifespan of
        the channel.

        """
        self.default_ticket = args.read_short()
        return self.default_ticket


    #############
    #
    #  Exchange
    #
    #
    # Exchanges match and distribute messages across queues.  Exchanges can be
    # configured in the server or created at runtime.
    #
    # GRAMMAR:
    #
    #     exchange            = C:DECLARE  S:DECLARE-OK
    #                         / C:DELETE   S:DELETE-OK
    #
    #

    def exchange_declare(self, exchange, type, passive=False, durable=False, auto_delete=False, internal=False, nowait=False, arguments={}, ticket=None):
        """
        This method creates an exchange if it does not already exist, and if the
        exchange exists, verifies that it is of the correct and expected class.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(exchange)
        args.write_shortstr(type)
        args.write_bit(passive)
        args.write_bit(durable)
        args.write_bit(auto_delete)
        args.write_bit(internal)
        args.write_bit(nowait)
        args.write_table(arguments)
        self._send_method_frame((40, 10), args)

        if not nowait:
            return self.wait(allowed_methods=[
                              (40, 11),    # Channel.exchange_declare_ok
                            ])


    def _exchange_declare_ok(self, args):
        """
        This method confirms a Declare method and confirms the name of the
        exchange, essential for automatically-named exchanges.

        """
        pass


    def exchange_delete(self, exchange, if_unused=False, nowait=False, ticket=None):
        """
        This method deletes an exchange.  When an exchange is deleted all queue
        bindings on the exchange are cancelled.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(exchange)
        args.write_bit(if_unused)
        args.write_bit(nowait)
        self._send_method_frame((40, 20), args)

        if not nowait:
            return self.wait(allowed_methods=[
                              (40, 21),    # Channel.exchange_delete_ok
                            ])


    def _exchange_delete_ok(self, args):
        """
        This method confirms the deletion of an exchange.

        """
        pass


    #############
    #
    #  Queue
    #
    #
    # Queues store and forward messages.  Queues can be configured in the server
    # or created at runtime.  Queues must be attached to at least one exchange
    # in order to receive messages from publishers.
    #
    # GRAMMAR:
    #
    #     queue               = C:DECLARE  S:DECLARE-OK
    #                         / C:BIND     S:BIND-OK
    #                         / C:PURGE    S:PURGE-OK
    #                         / C:DELETE   S:DELETE-OK
    #
    #

    def queue_bind(self, queue, exchange, routing_key='', nowait=False, arguments={}, ticket=None):
        """
        This method binds a queue to an exchange.  Until a queue is
        bound it will not receive any messages.  In a classic messaging
        model, store-and-forward queues are bound to a dest exchange
        and subscription queues are bound to a dest_wild exchange.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_shortstr(exchange)
        args.write_shortstr(routing_key)
        args.write_bit(nowait)
        args.write_table(arguments)
        self._send_method_frame((50, 20), args)

        if not nowait:
            return self.wait(allowed_methods=[
                              (50, 21),    # Channel.queue_bind_ok
                            ])


    def _queue_bind_ok(self, args):
        """
        This method confirms that the bind was successful.

        """
        pass


    def queue_declare(self, queue='', passive=False, durable=False, exclusive=False, auto_delete=False, nowait=False, arguments={}, ticket=None):
        """
        This method creates or checks a queue.  When creating a new queue
        the client can specify various properties that control the durability
        of the queue and its contents, and the level of sharing for the queue.

        Returns a tuple containing 3 items:
            the name of the queue (essential for automatically-named queues)
            message count
            consumer count

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_bit(passive)
        args.write_bit(durable)
        args.write_bit(exclusive)
        args.write_bit(auto_delete)
        args.write_bit(nowait)
        args.write_table(arguments)
        self._send_method_frame((50, 10), args)

        if not nowait:
            return self.wait(allowed_methods=[
                              (50, 11),    # Channel.queue_declare_ok
                            ])


    def _queue_declare_ok(self, args):
        """
        This method confirms a Declare method and confirms the name of the
        queue, essential for automatically-named queues.

        """
        queue = args.read_shortstr()
        message_count = args.read_long()
        consumer_count = args.read_long()

        return queue, message_count, consumer_count


    def queue_delete(self, queue, if_unused=False, if_empty=False, nowait=False, ticket=None):
        """
        This method deletes a queue.  When a queue is deleted any pending
        messages are sent to a dead-letter queue if this is defined in the
        server configuration, and all consumers on the queue are cancelled.

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_bit(if_unused)
        args.write_bit(if_empty)
        args.write_bit(nowait)
        self._send_method_frame((50, 40), args)

        if not nowait:
            return self.wait(allowed_methods=[
                              (50, 41),    # Channel.queue_delete_ok
                            ])


    def _queue_delete_ok(self, args):
        """
        This method confirms the deletion of a queue.

        """
        return args.read_long()


    def queue_purge(self, queue, nowait=False, ticket=None):
        """
        This method removes all messages from a queue.  It does not cancel
        consumers.  Purged messages are deleted without any formal "undo"
        mechanism.

        if nowait is False, returns a message_count

        """
        args = AMQPWriter()
        args.write_short(ticket if ticket is not None else self.default_ticket)
        args.write_shortstr(queue)
        args.write_bit(nowait)
        self._send_method_frame((50, 30), args)

        if not nowait:
            return self.wait(allowed_methods=[
                              (50, 31),    # Channel.queue_purge_ok
                            ])


    def _queue_purge_ok(self, args):
        """
        This method confirms the purge of a queue.

        """
        return args.read_long()


    #############
    #
    #  Basic
    #
    #
    # The Basic class provides methods that support an industry-standard
    # messaging model.
    #
    # GRAMMAR:
    #
    #     basic               = C:QOS S:QOS-OK
    #                         / C:CONSUME S:CONSUME-OK
    #                         / C:CANCEL S:CANCEL-OK
    #                         / C:PUBLISH content
    #                         / S:RETURN content
    #                         / S:DELIVER content
    #                         / C:GET ( S:GET-OK content / S:GET-EMPTY )
    #                         / C:ACK
    #                         / C:REJECT
{'fraction_non_alphanumeric': 0.0846312484609702, 'fraction_numerical': 0.007633587786259542, 'mean_word_length': 2.0008312935851844, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2602371', 'n_tokens_mistral': 18605, 'n_tokens_neox': 17525, 'n_words': 5477}