Merge lp:~xianghui/ubuntu/trusty/oslo.messaging/juno-lp1318721 into lp:~ubuntu-cloud-archive/ubuntu/trusty/oslo.messaging/juno

[Xiang Hui]

  * Backport of upstream release. (LP: #1318721):-
    - d/p/0007-fix-reconnect-race-condition-with-rabbitmq-cluster.patch:
      Redeclare if exception is catched after self.queue.declare() failed.

Unmerged revisions

3. By Xiang Hui on 2015-12-17

  * Backport of upstream release. (LP: #1318721):-
    - d/p/0007-fix-reconnect-race-condition-with-rabbitmq-cluster.patch:
      Redeclare if exception is catched after self.queue.declare() failed.

=== added directory '.pc'
=== added file '.pc/.quilt_patches'
--- .pc/.quilt_patches	1970-01-01 00:00:00 +0000
+++ .pc/.quilt_patches	2015-12-17 11:03:34 +0000
@@ -0,0 +1,1 @@
+debian/patches
 
=== added file '.pc/.quilt_series'
--- .pc/.quilt_series	1970-01-01 00:00:00 +0000
+++ .pc/.quilt_series	2015-12-17 11:03:34 +0000
@@ -0,0 +1,1 @@
+series
 
=== added file '.pc/.version'
--- .pc/.version	1970-01-01 00:00:00 +0000
+++ .pc/.version	2015-12-17 11:03:34 +0000
@@ -0,0 +1,1 @@
+2
 
=== added directory '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch'
=== added file '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/.timestamp'
=== added directory '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo'
=== added directory '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging'
=== added directory '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers'
=== added file '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/amqpdriver.py'
--- .pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/amqpdriver.py	1970-01-01 00:00:00 +0000
+++ .pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/amqpdriver.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,444 @@
+
+# Copyright 2013 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+__all__ = ['AMQPDriverBase']
+
+import logging
+import threading
+import time
+import uuid
+
+from six import moves
+
+from oslo import messaging
+from oslo.messaging._drivers import amqp as rpc_amqp
+from oslo.messaging._drivers import base
+from oslo.messaging._drivers import common as rpc_common
+
+LOG = logging.getLogger(__name__)
+
+
+class AMQPIncomingMessage(base.IncomingMessage):
+
+    def __init__(self, listener, ctxt, message, unique_id, msg_id, reply_q):
+        super(AMQPIncomingMessage, self).__init__(listener, ctxt,
+                                                  dict(message))
+
+        self.unique_id = unique_id
+        self.msg_id = msg_id
+        self.reply_q = reply_q
+        self.acknowledge_callback = message.acknowledge
+        self.requeue_callback = message.requeue
+
+    def _send_reply(self, conn, reply=None, failure=None,
+                    ending=False, log_failure=True):
+        if failure:
+            failure = rpc_common.serialize_remote_exception(failure,
+                                                            log_failure)
+
+        msg = {'result': reply, 'failure': failure}
+        if ending:
+            msg['ending'] = True
+
+        rpc_amqp._add_unique_id(msg)
+
+        # If a reply_q exists, add the msg_id to the reply and pass the
+        # reply_q to direct_send() to use it as the response queue.
+        # Otherwise use the msg_id for backward compatibility.
+        if self.reply_q:
+            msg['_msg_id'] = self.msg_id
+            conn.direct_send(self.reply_q, rpc_common.serialize_msg(msg))
+        else:
+            conn.direct_send(self.msg_id, rpc_common.serialize_msg(msg))
+
+    def reply(self, reply=None, failure=None, log_failure=True):
+        if not self.msg_id:
+            # NOTE(Alexei_987) not sending reply, if msg_id is empty
+            #    because reply should not be expected by caller side
+            return
+        with self.listener.driver._get_connection() as conn:
+            self._send_reply(conn, reply, failure, log_failure=log_failure)
+            self._send_reply(conn, ending=True)
+
+    def acknowledge(self):
+        self.listener.msg_id_cache.add(self.unique_id)
+        self.acknowledge_callback()
+
+    def requeue(self):
+        # NOTE(sileht): In case of the connection is lost between receiving the
+        # message and requeing it, this requeue call fail
+        # but because the message is not acknowledged and not added to the
+        # msg_id_cache, the message will be reconsumed, the only difference is
+        # the message stay at the beginning of the queue instead of moving to
+        # the end.
+        self.requeue_callback()
+
+
+class AMQPListener(base.Listener):
+
+    def __init__(self, driver, conn):
+        super(AMQPListener, self).__init__(driver)
+        self.conn = conn
+        self.msg_id_cache = rpc_amqp._MsgIdCache()
+        self.incoming = []
+
+    def __call__(self, message):
+        # FIXME(markmc): logging isn't driver specific
+        rpc_common._safe_log(LOG.debug, 'received %s', dict(message))
+
+        unique_id = self.msg_id_cache.check_duplicate_message(message)
+        ctxt = rpc_amqp.unpack_context(self.conf, message)
+
+        self.incoming.append(AMQPIncomingMessage(self,
+                                                 ctxt.to_dict(),
+                                                 message,
+                                                 unique_id,
+                                                 ctxt.msg_id,
+                                                 ctxt.reply_q))
+
+    def poll(self, timeout=None):
+        if timeout is not None:
+            deadline = time.time() + timeout
+        else:
+            deadline = None
+        while True:
+            if self.incoming:
+                return self.incoming.pop(0)
+            if deadline is not None:
+                timeout = deadline - time.time()
+                if timeout < 0:
+                    return None
+                try:
+                    self.conn.consume(limit=1, timeout=timeout)
+                except rpc_common.Timeout:
+                    return None
+            else:
+                self.conn.consume(limit=1)
+
+
+class ReplyWaiters(object):
+
+    WAKE_UP = object()
+
+    def __init__(self):
+        self._queues = {}
+        self._wrn_threshold = 10
+
+    def get(self, msg_id, timeout):
+        try:
+            return self._queues[msg_id].get(block=True, timeout=timeout)
+        except moves.queue.Empty:
+            raise messaging.MessagingTimeout('Timed out waiting for a reply '
+                                             'to message ID %s' % msg_id)
+
+    def check(self, msg_id):
+        try:
+            return self._queues[msg_id].get(block=False)
+        except moves.queue.Empty:
+            return None
+
+    def put(self, msg_id, message_data):
+        queue = self._queues.get(msg_id)
+        if not queue:
+            LOG.warn('No calling threads waiting for msg_id : %(msg_id)s'
+                     ', message : %(data)s', {'msg_id': msg_id,
+                                              'data': message_data})
+            LOG.warn('_queues: %s', self._queues)
+        else:
+            queue.put(message_data)
+
+    def wake_all(self, except_id):
+        msg_ids = [i for i in self._queues.keys() if i != except_id]
+        for msg_id in msg_ids:
+            self.put(msg_id, self.WAKE_UP)
+
+    def add(self, msg_id, queue):
+        self._queues[msg_id] = queue
+        if len(self._queues) > self._wrn_threshold:
+            LOG.warn('Number of call queues is greater than warning '
+                     'threshold: %d. There could be a leak.',
+                     self._wrn_threshold)
+            self._wrn_threshold *= 2
+
+    def remove(self, msg_id):
+        del self._queues[msg_id]
+
+
+class ReplyWaiter(object):
+
+    def __init__(self, conf, reply_q, conn, allowed_remote_exmods):
+        self.conf = conf
+        self.conn = conn
+        self.reply_q = reply_q
+        self.allowed_remote_exmods = allowed_remote_exmods
+
+        self.conn_lock = threading.Lock()
+        self.incoming = []
+        self.msg_id_cache = rpc_amqp._MsgIdCache()
+        self.waiters = ReplyWaiters()
+
+        conn.declare_direct_consumer(reply_q, self)
+
+    def __call__(self, message):
+        message.acknowledge()
+        self.incoming.append(message)
+
+    def listen(self, msg_id):
+        queue = moves.queue.Queue()
+        self.waiters.add(msg_id, queue)
+
+    def unlisten(self, msg_id):
+        self.waiters.remove(msg_id)
+
+    def _process_reply(self, data):
+        result = None
+        ending = False
+        self.msg_id_cache.check_duplicate_message(data)
+        if data['failure']:
+            failure = data['failure']
+            result = rpc_common.deserialize_remote_exception(
+                failure, self.allowed_remote_exmods)
+        elif data.get('ending', False):
+            ending = True
+        else:
+            result = data['result']
+        return result, ending
+
+    def _poll_connection(self, msg_id, timeout):
+        while True:
+            while self.incoming:
+                message_data = self.incoming.pop(0)
+
+                incoming_msg_id = message_data.pop('_msg_id', None)
+                if incoming_msg_id == msg_id:
+                    return self._process_reply(message_data)
+
+                self.waiters.put(incoming_msg_id, message_data)
+
+            try:
+                self.conn.consume(limit=1, timeout=timeout)
+            except rpc_common.Timeout:
+                raise messaging.MessagingTimeout('Timed out waiting for a '
+                                                 'reply to message ID %s'
+                                                 % msg_id)
+
+    def _poll_queue(self, msg_id, timeout):
+        message = self.waiters.get(msg_id, timeout)
+        if message is self.waiters.WAKE_UP:
+            return None, None, True  # lock was released
+
+        reply, ending = self._process_reply(message)
+        return reply, ending, False
+
+    def _check_queue(self, msg_id):
+        while True:
+            message = self.waiters.check(msg_id)
+            if message is self.waiters.WAKE_UP:
+                continue
+            if message is None:
+                return None, None, True  # queue is empty
+
+            reply, ending = self._process_reply(message)
+            return reply, ending, False
+
+    def wait(self, msg_id, timeout):
+        #
+        # NOTE(markmc): we're waiting for a reply for msg_id to come in for on
+        # the reply_q, but there may be other threads also waiting for replies
+        # to other msg_ids
+        #
+        # Only one thread can be consuming from the queue using this connection
+        # and we don't want to hold open a connection per thread, so instead we
+        # have the first thread take responsibility for passing replies not
+        # intended for itself to the appropriate thread.
+        #
+        final_reply = None
+        while True:
+            if self.conn_lock.acquire(False):
+                # Ok, we're the thread responsible for polling the connection
+                try:
+                    # Check the queue to see if a previous lock-holding thread
+                    # queued up a reply already
+                    while True:
+                        reply, ending, empty = self._check_queue(msg_id)
+                        if empty:
+                            break
+                        if not ending:
+                            final_reply = reply
+                        else:
+                            return final_reply
+
+                    # Now actually poll the connection
+                    while True:
+                        reply, ending = self._poll_connection(msg_id, timeout)
+                        if not ending:
+                            final_reply = reply
+                        else:
+                            return final_reply
+                finally:
+                    self.conn_lock.release()
+                    # We've got our reply, tell the other threads to wake up
+                    # so that one of them will take over the responsibility for
+                    # polling the connection
+                    self.waiters.wake_all(msg_id)
+            else:
+                # We're going to wait for the first thread to pass us our reply
+                reply, ending, trylock = self._poll_queue(msg_id, timeout)
+                if trylock:
+                    # The first thread got its reply, let's try and take over
+                    # the responsibility for polling
+                    continue
+                if not ending:
+                    final_reply = reply
+                else:
+                    return final_reply
+
+
+class AMQPDriverBase(base.BaseDriver):
+
+    def __init__(self, conf, url, connection_pool,
+                 default_exchange=None, allowed_remote_exmods=None):
+        super(AMQPDriverBase, self).__init__(conf, url, default_exchange,
+                                             allowed_remote_exmods)
+
+        self._default_exchange = default_exchange
+
+        self._connection_pool = connection_pool
+
+        self._reply_q_lock = threading.Lock()
+        self._reply_q = None
+        self._reply_q_conn = None
+        self._waiter = None
+
+    def _get_exchange(self, target):
+        return target.exchange or self._default_exchange
+
+    def _get_connection(self, pooled=True):
+        return rpc_amqp.ConnectionContext(self.conf,
+                                          self._url,
+                                          self._connection_pool,
+                                          pooled=pooled)
+
+    def _get_reply_q(self):
+        with self._reply_q_lock:
+            if self._reply_q is not None:
+                return self._reply_q
+
+            reply_q = 'reply_' + uuid.uuid4().hex
+
+            conn = self._get_connection(pooled=False)
+
+            self._waiter = ReplyWaiter(self.conf, reply_q, conn,
+                                       self._allowed_remote_exmods)
+
+            self._reply_q = reply_q
+            self._reply_q_conn = conn
+
+        return self._reply_q
+
+    def _send(self, target, ctxt, message,
+              wait_for_reply=None, timeout=None,
+              envelope=True, notify=False, retry=None):
+
+        # FIXME(markmc): remove this temporary hack
+        class Context(object):
+            def __init__(self, d):
+                self.d = d
+
+            def to_dict(self):
+                return self.d
+
+        context = Context(ctxt)
+        msg = message
+
+        if wait_for_reply:
+            msg_id = uuid.uuid4().hex
+            msg.update({'_msg_id': msg_id})
+            LOG.debug('MSG_ID is %s', msg_id)
+            msg.update({'_reply_q': self._get_reply_q()})
+
+        rpc_amqp._add_unique_id(msg)
+        rpc_amqp.pack_context(msg, context)
+
+        if envelope:
+            msg = rpc_common.serialize_msg(msg)
+
+        if wait_for_reply:
+            self._waiter.listen(msg_id)
+
+        try:
+            with self._get_connection() as conn:
+                if notify:
+                    conn.notify_send(self._get_exchange(target),
+                                     target.topic, msg, retry=retry)
+                elif target.fanout:
+                    conn.fanout_send(target.topic, msg, retry=retry)
+                else:
+                    topic = target.topic
+                    if target.server:
+                        topic = '%s.%s' % (target.topic, target.server)
+                    conn.topic_send(exchange_name=self._get_exchange(target),
+                                    topic=topic, msg=msg, timeout=timeout,
+                                    retry=retry)
+
+            if wait_for_reply:
+                result = self._waiter.wait(msg_id, timeout)
+                if isinstance(result, Exception):
+                    raise result
+                return result
+        finally:
+            if wait_for_reply:
+                self._waiter.unlisten(msg_id)
+
+    def send(self, target, ctxt, message, wait_for_reply=None, timeout=None,
+             retry=None):
+        return self._send(target, ctxt, message, wait_for_reply, timeout,
+                          retry=retry)
+
+    def send_notification(self, target, ctxt, message, version, retry=None):
+        return self._send(target, ctxt, message,
+                          envelope=(version == 2.0), notify=True, retry=retry)
+
+    def listen(self, target):
+        conn = self._get_connection(pooled=False)
+
+        listener = AMQPListener(self, conn)
+
+        conn.declare_topic_consumer(exchange_name=self._get_exchange(target),
+                                    topic=target.topic,
+                                    callback=listener)
+        conn.declare_topic_consumer(exchange_name=self._get_exchange(target),
+                                    topic='%s.%s' % (target.topic,
+                                                     target.server),
+                                    callback=listener)
+        conn.declare_fanout_consumer(target.topic, listener)
+
+        return listener
+
+    def listen_for_notifications(self, targets_and_priorities):
+        conn = self._get_connection(pooled=False)
+
+        listener = AMQPListener(self, conn)
+        for target, priority in targets_and_priorities:
+            conn.declare_topic_consumer(
+                exchange_name=self._get_exchange(target),
+                topic='%s.%s' % (target.topic, priority),
+                callback=listener)
+        return listener
+
+    def cleanup(self):
+        if self._connection_pool:
+            self._connection_pool.empty()
+        self._connection_pool = None
 
=== added file '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/common.py'
--- .pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/common.py	1970-01-01 00:00:00 +0000
+++ .pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/common.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,349 @@
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+# Copyright 2011 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import copy
+import logging
+import sys
+import traceback
+
+import six
+
+from oslo import messaging
+from oslo.messaging import _utils as utils
+from oslo.messaging.openstack.common.gettextutils import _
+from oslo.messaging.openstack.common import jsonutils
+
+LOG = logging.getLogger(__name__)
+
+_EXCEPTIONS_MODULE = 'exceptions' if six.PY2 else 'builtins'
+
+
+'''RPC Envelope Version.
+
+This version number applies to the top level structure of messages sent out.
+It does *not* apply to the message payload, which must be versioned
+independently.  For example, when using rpc APIs, a version number is applied
+for changes to the API being exposed over rpc.  This version number is handled
+in the rpc proxy and dispatcher modules.
+
+This version number applies to the message envelope that is used in the
+serialization done inside the rpc layer.  See serialize_msg() and
+deserialize_msg().
+
+The current message format (version 2.0) is very simple.  It is:
+
+    {
+        'oslo.version': <RPC Envelope Version as a String>,
+        'oslo.message': <Application Message Payload, JSON encoded>
+    }
+
+Message format version '1.0' is just considered to be the messages we sent
+without a message envelope.
+
+So, the current message envelope just includes the envelope version.  It may
+eventually contain additional information, such as a signature for the message
+payload.
+
+We will JSON encode the application message payload.  The message envelope,
+which includes the JSON encoded application message body, will be passed down
+to the messaging libraries as a dict.
+'''
+_RPC_ENVELOPE_VERSION = '2.0'
+
+_VERSION_KEY = 'oslo.version'
+_MESSAGE_KEY = 'oslo.message'
+
+_REMOTE_POSTFIX = '_Remote'
+
+
+class RPCException(Exception):
+    msg_fmt = _("An unknown RPC related exception occurred.")
+
+    def __init__(self, message=None, **kwargs):
+        self.kwargs = kwargs
+
+        if not message:
+            try:
+                message = self.msg_fmt % kwargs
+
+            except Exception:
+                # kwargs doesn't match a variable in the message
+                # log the issue and the kwargs
+                LOG.exception(_('Exception in string format operation'))
+                for name, value in six.iteritems(kwargs):
+                    LOG.error("%s: %s", name, value)
+                # at least get the core message out if something happened
+                message = self.msg_fmt
+
+        super(RPCException, self).__init__(message)
+
+
+class Timeout(RPCException):
+    """Signifies that a timeout has occurred.
+
+    This exception is raised if the rpc_response_timeout is reached while
+    waiting for a response from the remote side.
+    """
+    msg_fmt = _('Timeout while waiting on RPC response - '
+                'topic: "%(topic)s", RPC method: "%(method)s" '
+                'info: "%(info)s"')
+
+    def __init__(self, info=None, topic=None, method=None):
+        """Initiates Timeout object.
+
+        :param info: Extra info to convey to the user
+        :param topic: The topic that the rpc call was sent to
+        :param rpc_method_name: The name of the rpc method being
+                                called
+        """
+        self.info = info
+        self.topic = topic
+        self.method = method
+        super(Timeout, self).__init__(
+            None,
+            info=info or _('<unknown>'),
+            topic=topic or _('<unknown>'),
+            method=method or _('<unknown>'))
+
+
+class DuplicateMessageError(RPCException):
+    msg_fmt = _("Found duplicate message(%(msg_id)s). Skipping it.")
+
+
+class InvalidRPCConnectionReuse(RPCException):
+    msg_fmt = _("Invalid reuse of an RPC connection.")
+
+
+class UnsupportedRpcVersion(RPCException):
+    msg_fmt = _("Specified RPC version, %(version)s, not supported by "
+                "this endpoint.")
+
+
+class UnsupportedRpcEnvelopeVersion(RPCException):
+    msg_fmt = _("Specified RPC envelope version, %(version)s, "
+                "not supported by this endpoint.")
+
+
+class RpcVersionCapError(RPCException):
+    msg_fmt = _("Specified RPC version cap, %(version_cap)s, is too low")
+
+
+class Connection(object):
+    """A connection, returned by rpc.create_connection().
+
+    This class represents a connection to the message bus used for rpc.
+    An instance of this class should never be created by users of the rpc API.
+    Use rpc.create_connection() instead.
+    """
+    def close(self):
+        """Close the connection.
+
+        This method must be called when the connection will no longer be used.
+        It will ensure that any resources associated with the connection, such
+        as a network connection, and cleaned up.
+        """
+        raise NotImplementedError()
+
+
+def _safe_log(log_func, msg, msg_data):
+    """Sanitizes the msg_data field before logging."""
+    SANITIZE = ['_context_auth_token', 'auth_token', 'new_pass']
+
+    def _fix_passwords(d):
+        """Sanitizes the password fields in the dictionary."""
+        for k in six.iterkeys(d):
+            if k.lower().find('password') != -1:
+                d[k] = '<SANITIZED>'
+            elif k.lower() in SANITIZE:
+                d[k] = '<SANITIZED>'
+            elif isinstance(d[k], dict):
+                _fix_passwords(d[k])
+        return d
+
+    return log_func(msg, _fix_passwords(copy.deepcopy(msg_data)))
+
+
+def serialize_remote_exception(failure_info, log_failure=True):
+    """Prepares exception data to be sent over rpc.
+
+    Failure_info should be a sys.exc_info() tuple.
+
+    """
+    tb = traceback.format_exception(*failure_info)
+    failure = failure_info[1]
+    if log_failure:
+        LOG.error(_("Returning exception %s to caller"),
+                  six.text_type(failure))
+        LOG.error(tb)
+
+    kwargs = {}
+    if hasattr(failure, 'kwargs'):
+        kwargs = failure.kwargs
+
+    # NOTE(matiu): With cells, it's possible to re-raise remote, remote
+    # exceptions. Lets turn it back into the original exception type.
+    cls_name = six.text_type(failure.__class__.__name__)
+    mod_name = six.text_type(failure.__class__.__module__)
+    if (cls_name.endswith(_REMOTE_POSTFIX) and
+            mod_name.endswith(_REMOTE_POSTFIX)):
+        cls_name = cls_name[:-len(_REMOTE_POSTFIX)]
+        mod_name = mod_name[:-len(_REMOTE_POSTFIX)]
+
+    data = {
+        'class': cls_name,
+        'module': mod_name,
+        'message': six.text_type(failure),
+        'tb': tb,
+        'args': failure.args,
+        'kwargs': kwargs
+    }
+
+    json_data = jsonutils.dumps(data)
+
+    return json_data
+
+
+def deserialize_remote_exception(data, allowed_remote_exmods):
+    failure = jsonutils.loads(six.text_type(data))
+
+    trace = failure.get('tb', [])
+    message = failure.get('message', "") + "\n" + "\n".join(trace)
+    name = failure.get('class')
+    module = failure.get('module')
+
+    # NOTE(ameade): We DO NOT want to allow just any module to be imported, in
+    # order to prevent arbitrary code execution.
+    if module != _EXCEPTIONS_MODULE and module not in allowed_remote_exmods:
+        return messaging.RemoteError(name, failure.get('message'), trace)
+
+    try:
+        __import__(module)
+        mod = sys.modules[module]
+        klass = getattr(mod, name)
+        if not issubclass(klass, Exception):
+            raise TypeError("Can only deserialize Exceptions")
+
+        failure = klass(*failure.get('args', []), **failure.get('kwargs', {}))
+    except (AttributeError, TypeError, ImportError):
+        return messaging.RemoteError(name, failure.get('message'), trace)
+
+    ex_type = type(failure)
+    str_override = lambda self: message
+    new_ex_type = type(ex_type.__name__ + _REMOTE_POSTFIX, (ex_type,),
+                       {'__str__': str_override, '__unicode__': str_override})
+    new_ex_type.__module__ = '%s%s' % (module, _REMOTE_POSTFIX)
+    try:
+        # NOTE(ameade): Dynamically create a new exception type and swap it in
+        # as the new type for the exception. This only works on user defined
+        # Exceptions and not core Python exceptions. This is important because
+        # we cannot necessarily change an exception message so we must override
+        # the __str__ method.
+        failure.__class__ = new_ex_type
+    except TypeError:
+        # NOTE(ameade): If a core exception then just add the traceback to the
+        # first exception argument.
+        failure.args = (message,) + failure.args[1:]
+    return failure
+
+
+class CommonRpcContext(object):
+    def __init__(self, **kwargs):
+        self.values = kwargs
+
+    def __getattr__(self, key):
+        try:
+            return self.values[key]
+        except KeyError:
+            raise AttributeError(key)
+
+    def to_dict(self):
+        return copy.deepcopy(self.values)
+
+    @classmethod
+    def from_dict(cls, values):
+        return cls(**values)
+
+    def deepcopy(self):
+        return self.from_dict(self.to_dict())
+
+    def update_store(self):
+        # local.store.context = self
+        pass
+
+
+class ClientException(Exception):
+    """Encapsulates actual exception expected to be hit by a RPC proxy object.
+
+    Merely instantiating it records the current exception information, which
+    will be passed back to the RPC client without exceptional logging.
+    """
+    def __init__(self):
+        self._exc_info = sys.exc_info()
+
+
+def serialize_msg(raw_msg):
+    # NOTE(russellb) See the docstring for _RPC_ENVELOPE_VERSION for more
+    # information about this format.
+    msg = {_VERSION_KEY: _RPC_ENVELOPE_VERSION,
+           _MESSAGE_KEY: jsonutils.dumps(raw_msg)}
+
+    return msg
+
+
+def deserialize_msg(msg):
+    # NOTE(russellb): Hang on to your hats, this road is about to
+    # get a little bumpy.
+    #
+    # Robustness Principle:
+    #    "Be strict in what you send, liberal in what you accept."
+    #
+    # At this point we have to do a bit of guessing about what it
+    # is we just received.  Here is the set of possibilities:
+    #
+    # 1) We received a dict.  This could be 2 things:
+    #
+    #   a) Inspect it to see if it looks like a standard message envelope.
+    #      If so, great!
+    #
+    #   b) If it doesn't look like a standard message envelope, it could either
+    #      be a notification, or a message from before we added a message
+    #      envelope (referred to as version 1.0).
+    #      Just return the message as-is.
+    #
+    # 2) It's any other non-dict type.  Just return it and hope for the best.
+    #    This case covers return values from rpc.call() from before message
+    #    envelopes were used.  (messages to call a method were always a dict)
+
+    if not isinstance(msg, dict):
+        # See #2 above.
+        return msg
+
+    base_envelope_keys = (_VERSION_KEY, _MESSAGE_KEY)
+    if not all(map(lambda key: key in msg, base_envelope_keys)):
+        #  See #1.b above.
+        return msg
+
+    # At this point we think we have the message envelope
+    # format we were expecting. (#1.a above)
+
+    if not utils.version_is_compatible(_RPC_ENVELOPE_VERSION,
+                                       msg[_VERSION_KEY]):
+        raise UnsupportedRpcEnvelopeVersion(version=msg[_VERSION_KEY])
+
+    raw_msg = jsonutils.loads(msg[_MESSAGE_KEY])
+
+    return raw_msg
 
=== added file '.pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/impl_rabbit.py'
--- .pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/impl_rabbit.py	1970-01-01 00:00:00 +0000
+++ .pc/0001-Fix-_poll_connection-not-timeout-issue-1-2.patch/oslo/messaging/_drivers/impl_rabbit.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,821 @@
+#    Copyright 2011 OpenStack Foundation
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+import itertools
+import logging
+import random
+import socket
+import ssl
+import time
+import uuid
+
+import kombu
+import kombu.connection
+import kombu.entity
+import kombu.messaging
+import six
+
+from oslo.config import cfg
+from oslo.messaging._drivers import amqp as rpc_amqp
+from oslo.messaging._drivers import amqpdriver
+from oslo.messaging._drivers import common as rpc_common
+from oslo.messaging import exceptions
+from oslo.messaging.openstack.common.gettextutils import _
+from oslo.utils import netutils
+
+rabbit_opts = [
+    cfg.StrOpt('kombu_ssl_version',
+               default='',
+               help='SSL version to use (valid only if SSL enabled). '
+                    'valid values are TLSv1, SSLv23 and SSLv3. SSLv2 may '
+                    'be available on some distributions.'
+               ),
+    cfg.StrOpt('kombu_ssl_keyfile',
+               default='',
+               help='SSL key file (valid only if SSL enabled).'),
+    cfg.StrOpt('kombu_ssl_certfile',
+               default='',
+               help='SSL cert file (valid only if SSL enabled).'),
+    cfg.StrOpt('kombu_ssl_ca_certs',
+               default='',
+               help='SSL certification authority file '
+                    '(valid only if SSL enabled).'),
+    cfg.FloatOpt('kombu_reconnect_delay',
+                 default=1.0,
+                 help='How long to wait before reconnecting in response to an '
+                      'AMQP consumer cancel notification.'),
+    cfg.StrOpt('rabbit_host',
+               default='localhost',
+               help='The RabbitMQ broker address where a single node is '
+                    'used.'),
+    cfg.IntOpt('rabbit_port',
+               default=5672,
+               help='The RabbitMQ broker port where a single node is used.'),
+    cfg.ListOpt('rabbit_hosts',
+                default=['$rabbit_host:$rabbit_port'],
+                help='RabbitMQ HA cluster host:port pairs.'),
+    cfg.BoolOpt('rabbit_use_ssl',
+                default=False,
+                help='Connect over SSL for RabbitMQ.'),
+    cfg.StrOpt('rabbit_userid',
+               default='guest',
+               help='The RabbitMQ userid.'),
+    cfg.StrOpt('rabbit_password',
+               default='guest',
+               help='The RabbitMQ password.',
+               secret=True),
+    cfg.StrOpt('rabbit_login_method',
+               default='AMQPLAIN',
+               help='the RabbitMQ login method'),
+    cfg.StrOpt('rabbit_virtual_host',
+               default='/',
+               help='The RabbitMQ virtual host.'),
+    cfg.IntOpt('rabbit_retry_interval',
+               default=1,
+               help='How frequently to retry connecting with RabbitMQ.'),
+    cfg.IntOpt('rabbit_retry_backoff',
+               default=2,
+               help='How long to backoff for between retries when connecting '
+                    'to RabbitMQ.'),
+    cfg.IntOpt('rabbit_max_retries',
+               default=0,
+               help='Maximum number of RabbitMQ connection retries. '
+                    'Default is 0 (infinite retry count).'),
+    cfg.BoolOpt('rabbit_ha_queues',
+                default=False,
+                help='Use HA queues in RabbitMQ (x-ha-policy: all). '
+                     'If you change this option, you must wipe the '
+                     'RabbitMQ database.'),
+
+    # FIXME(markmc): this was toplevel in openstack.common.rpc
+    cfg.BoolOpt('fake_rabbit',
+                default=False,
+                help='If passed, use a fake RabbitMQ provider.'),
+]
+
+LOG = logging.getLogger(__name__)
+
+
+def _get_queue_arguments(conf):
+    """Construct the arguments for declaring a queue.
+
+    If the rabbit_ha_queues option is set, we declare a mirrored queue
+    as described here:
+
+      http://www.rabbitmq.com/ha.html
+
+    Setting x-ha-policy to all means that the queue will be mirrored
+    to all nodes in the cluster.
+    """
+    return {'x-ha-policy': 'all'} if conf.rabbit_ha_queues else {}
+
+
+class RabbitMessage(dict):
+    def __init__(self, raw_message):
+        super(RabbitMessage, self).__init__(
+            rpc_common.deserialize_msg(raw_message.payload))
+        self._raw_message = raw_message
+
+    def acknowledge(self):
+        self._raw_message.ack()
+
+    def requeue(self):
+        self._raw_message.requeue()
+
+
+class ConsumerBase(object):
+    """Consumer base class."""
+
+    def __init__(self, channel, callback, tag, **kwargs):
+        """Declare a queue on an amqp channel.
+
+        'channel' is the amqp channel to use
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        queue name, exchange name, and other kombu options are
+        passed in here as a dictionary.
+        """
+        self.callback = callback
+        self.tag = six.text_type(tag)
+        self.kwargs = kwargs
+        self.queue = None
+        self.reconnect(channel)
+
+    def reconnect(self, channel):
+        """Re-declare the queue after a rabbit reconnect."""
+        self.channel = channel
+        self.kwargs['channel'] = channel
+        self.queue = kombu.entity.Queue(**self.kwargs)
+        self.queue.declare()
+
+    def _callback_handler(self, message, callback):
+        """Call callback with deserialized message.
+
+        Messages that are processed and ack'ed.
+        """
+
+        try:
+            callback(RabbitMessage(message))
+        except Exception:
+            LOG.exception(_("Failed to process message"
+                            " ... skipping it."))
+            message.ack()
+
+    def consume(self, *args, **kwargs):
+        """Actually declare the consumer on the amqp channel.  This will
+        start the flow of messages from the queue.  Using the
+        Connection.iterconsume() iterator will process the messages,
+        calling the appropriate callback.
+
+        If a callback is specified in kwargs, use that.  Otherwise,
+        use the callback passed during __init__()
+
+        If kwargs['nowait'] is True, then this call will block until
+        a message is read.
+
+        """
+
+        options = {'consumer_tag': self.tag}
+        options['nowait'] = kwargs.get('nowait', False)
+        callback = kwargs.get('callback', self.callback)
+        if not callback:
+            raise ValueError("No callback defined")
+
+        def _callback(raw_message):
+            message = self.channel.message_to_python(raw_message)
+            self._callback_handler(message, callback)
+
+        self.queue.consume(*args, callback=_callback, **options)
+
+    def cancel(self):
+        """Cancel the consuming from the queue, if it has started."""
+        try:
+            self.queue.cancel(self.tag)
+        except KeyError as e:
+            # NOTE(comstud): Kludge to get around a amqplib bug
+            if six.text_type(e) != "u'%s'" % self.tag:
+                raise
+        self.queue = None
+
+
+class DirectConsumer(ConsumerBase):
+    """Queue/consumer class for 'direct'."""
+
+    def __init__(self, conf, channel, msg_id, callback, tag, **kwargs):
+        """Init a 'direct' queue.
+
+        'channel' is the amqp channel to use
+        'msg_id' is the msg_id to listen on
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        Other kombu options may be passed
+        """
+        # Default options
+        options = {'durable': False,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=msg_id,
+                                         type='direct',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(DirectConsumer, self).__init__(channel,
+                                             callback,
+                                             tag,
+                                             name=msg_id,
+                                             exchange=exchange,
+                                             routing_key=msg_id,
+                                             **options)
+
+
+class TopicConsumer(ConsumerBase):
+    """Consumer class for 'topic'."""
+
+    def __init__(self, conf, channel, topic, callback, tag, exchange_name,
+                 name=None, **kwargs):
+        """Init a 'topic' queue.
+
+        :param channel: the amqp channel to use
+        :param topic: the topic to listen on
+        :paramtype topic: str
+        :param callback: the callback to call when messages are received
+        :param tag: a unique ID for the consumer on the channel
+        :param exchange_name: the exchange name to use
+        :param name: optional queue name, defaults to topic
+        :paramtype name: str
+
+        Other kombu options may be passed as keyword arguments
+        """
+        # Default options
+        options = {'durable': conf.amqp_durable_queues,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': conf.amqp_auto_delete,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=exchange_name,
+                                         type='topic',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(TopicConsumer, self).__init__(channel,
+                                            callback,
+                                            tag,
+                                            name=name or topic,
+                                            exchange=exchange,
+                                            routing_key=topic,
+                                            **options)
+
+
+class FanoutConsumer(ConsumerBase):
+    """Consumer class for 'fanout'."""
+
+    def __init__(self, conf, channel, topic, callback, tag, **kwargs):
+        """Init a 'fanout' queue.
+
+        'channel' is the amqp channel to use
+        'topic' is the topic to listen on
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        Other kombu options may be passed
+        """
+        unique = uuid.uuid4().hex
+        exchange_name = '%s_fanout' % topic
+        queue_name = '%s_fanout_%s' % (topic, unique)
+
+        # Default options
+        options = {'durable': False,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=exchange_name, type='fanout',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(FanoutConsumer, self).__init__(channel, callback, tag,
+                                             name=queue_name,
+                                             exchange=exchange,
+                                             routing_key=topic,
+                                             **options)
+
+
+class Publisher(object):
+    """Base Publisher class."""
+
+    def __init__(self, channel, exchange_name, routing_key, **kwargs):
+        """Init the Publisher class with the exchange_name, routing_key,
+        and other options
+        """
+        self.exchange_name = exchange_name
+        self.routing_key = routing_key
+        self.kwargs = kwargs
+        self.reconnect(channel)
+
+    def reconnect(self, channel):
+        """Re-establish the Producer after a rabbit reconnection."""
+        self.exchange = kombu.entity.Exchange(name=self.exchange_name,
+                                              **self.kwargs)
+        self.producer = kombu.messaging.Producer(exchange=self.exchange,
+                                                 channel=channel,
+                                                 routing_key=self.routing_key)
+
+    def send(self, msg, timeout=None):
+        """Send a message."""
+        if timeout:
+            #
+            # AMQP TTL is in milliseconds when set in the header.
+            #
+            self.producer.publish(msg, headers={'ttl': (timeout * 1000)})
+        else:
+            self.producer.publish(msg)
+
+
+class DirectPublisher(Publisher):
+    """Publisher class for 'direct'."""
+    def __init__(self, conf, channel, topic, **kwargs):
+        """Init a 'direct' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+
+        options = {'durable': False,
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(DirectPublisher, self).__init__(channel, topic, topic,
+                                              type='direct', **options)
+
+
+class TopicPublisher(Publisher):
+    """Publisher class for 'topic'."""
+    def __init__(self, conf, channel, exchange_name, topic, **kwargs):
+        """Init a 'topic' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+        options = {'durable': conf.amqp_durable_queues,
+                   'auto_delete': conf.amqp_auto_delete,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(TopicPublisher, self).__init__(channel,
+                                             exchange_name,
+                                             topic,
+                                             type='topic',
+                                             **options)
+
+
+class FanoutPublisher(Publisher):
+    """Publisher class for 'fanout'."""
+    def __init__(self, conf, channel, topic, **kwargs):
+        """Init a 'fanout' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+        options = {'durable': False,
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(FanoutPublisher, self).__init__(channel, '%s_fanout' % topic,
+                                              None, type='fanout', **options)
+
+
+class NotifyPublisher(TopicPublisher):
+    """Publisher class for 'notify'."""
+
+    def __init__(self, conf, channel, exchange_name, topic, **kwargs):
+        self.durable = kwargs.pop('durable', conf.amqp_durable_queues)
+        self.queue_arguments = _get_queue_arguments(conf)
+        super(NotifyPublisher, self).__init__(conf, channel, exchange_name,
+                                              topic, **kwargs)
+
+    def reconnect(self, channel):
+        super(NotifyPublisher, self).reconnect(channel)
+
+        # NOTE(jerdfelt): Normally the consumer would create the queue, but
+        # we do this to ensure that messages don't get dropped if the
+        # consumer is started after we do
+        queue = kombu.entity.Queue(channel=channel,
+                                   exchange=self.exchange,
+                                   durable=self.durable,
+                                   name=self.routing_key,
+                                   routing_key=self.routing_key,
+                                   queue_arguments=self.queue_arguments)
+        queue.declare()
+
+
+class Connection(object):
+    """Connection object."""
+
+    pools = {}
+
+    def __init__(self, conf, url):
+        self.consumers = []
+        self.conf = conf
+        self.max_retries = self.conf.rabbit_max_retries
+        # Try forever?
+        if self.max_retries <= 0:
+            self.max_retries = None
+        self.interval_start = self.conf.rabbit_retry_interval
+        self.interval_stepping = self.conf.rabbit_retry_backoff
+        # max retry-interval = 30 seconds
+        self.interval_max = 30
+        self.memory_transport = False
+
+        ssl_params = self._fetch_ssl_params()
+
+        if url.virtual_host is not None:
+            virtual_host = url.virtual_host
+        else:
+            virtual_host = self.conf.rabbit_virtual_host
+
+        self.brokers_params = []
+        if url.hosts:
+            for host in url.hosts:
+                params = {
+                    'hostname': host.hostname,
+                    'port': host.port or 5672,
+                    'userid': host.username or '',
+                    'password': host.password or '',
+                    'login_method': self.conf.rabbit_login_method,
+                    'virtual_host': virtual_host
+                }
+                if self.conf.fake_rabbit:
+                    params['transport'] = 'memory'
+                if self.conf.rabbit_use_ssl:
+                    params['ssl'] = ssl_params
+
+                self.brokers_params.append(params)
+        else:
+            # Old configuration format
+            for adr in self.conf.rabbit_hosts:
+                hostname, port = netutils.parse_host_port(
+                    adr, default_port=self.conf.rabbit_port)
+
+                params = {
+                    'hostname': hostname,
+                    'port': port,
+                    'userid': self.conf.rabbit_userid,
+                    'password': self.conf.rabbit_password,
+                    'login_method': self.conf.rabbit_login_method,
+                    'virtual_host': virtual_host
+                }
+
+                if self.conf.fake_rabbit:
+                    params['transport'] = 'memory'
+                if self.conf.rabbit_use_ssl:
+                    params['ssl'] = ssl_params
+
+                self.brokers_params.append(params)
+
+        random.shuffle(self.brokers_params)
+        self.brokers = itertools.cycle(self.brokers_params)
+
+        self.memory_transport = self.conf.fake_rabbit
+
+        self.connection = None
+        self.do_consume = None
+        self.reconnect()
+
+    # FIXME(markmc): use oslo sslutils when it is available as a library
+    _SSL_PROTOCOLS = {
+        "tlsv1": ssl.PROTOCOL_TLSv1,
+        "sslv23": ssl.PROTOCOL_SSLv23,
+        "sslv3": ssl.PROTOCOL_SSLv3
+    }
+
+    try:
+        _SSL_PROTOCOLS["sslv2"] = ssl.PROTOCOL_SSLv2
+    except AttributeError:
+        pass
+
+    @classmethod
+    def validate_ssl_version(cls, version):
+        key = version.lower()
+        try:
+            return cls._SSL_PROTOCOLS[key]
+        except KeyError:
+            raise RuntimeError(_("Invalid SSL version : %s") % version)
+
+    def _fetch_ssl_params(self):
+        """Handles fetching what ssl params should be used for the connection
+        (if any).
+        """
+        ssl_params = dict()
+
+        # http://docs.python.org/library/ssl.html - ssl.wrap_socket
+        if self.conf.kombu_ssl_version:
+            ssl_params['ssl_version'] = self.validate_ssl_version(
+                self.conf.kombu_ssl_version)
+        if self.conf.kombu_ssl_keyfile:
+            ssl_params['keyfile'] = self.conf.kombu_ssl_keyfile
+        if self.conf.kombu_ssl_certfile:
+            ssl_params['certfile'] = self.conf.kombu_ssl_certfile
+        if self.conf.kombu_ssl_ca_certs:
+            ssl_params['ca_certs'] = self.conf.kombu_ssl_ca_certs
+            # We might want to allow variations in the
+            # future with this?
+            ssl_params['cert_reqs'] = ssl.CERT_REQUIRED
+
+        # Return the extended behavior or just have the default behavior
+        return ssl_params or True
+
+    def _connect(self, broker):
+        """Connect to rabbit.  Re-establish any queues that may have
+        been declared before if we are reconnecting.  Exceptions should
+        be handled by the caller.
+        """
+        LOG.info(_("Connecting to AMQP server on "
+                   "%(hostname)s:%(port)d"), broker)
+        self.connection = kombu.connection.BrokerConnection(**broker)
+        self.connection_errors = self.connection.connection_errors
+        self.channel_errors = self.connection.channel_errors
+        if self.memory_transport:
+            # Kludge to speed up tests.
+            self.connection.transport.polling_interval = 0.0
+        self.do_consume = True
+        self.consumer_num = itertools.count(1)
+        self.connection.connect()
+        self.channel = self.connection.channel()
+        # work around 'memory' transport bug in 1.1.3
+        if self.memory_transport:
+            self.channel._new_queue('ae.undeliver')
+        for consumer in self.consumers:
+            consumer.reconnect(self.channel)
+        LOG.info(_('Connected to AMQP server on %(hostname)s:%(port)d'),
+                 broker)
+
+    def _disconnect(self):
+        if self.connection:
+            # XXX(nic): when reconnecting to a RabbitMQ cluster
+            # with mirrored queues in use, the attempt to release the
+            # connection can hang "indefinitely" somewhere deep down
+            # in Kombu.  Blocking the thread for a bit prior to
+            # release seems to kludge around the problem where it is
+            # otherwise reproduceable.
+            if self.conf.kombu_reconnect_delay > 0:
+                LOG.info(_("Delaying reconnect for %1.1f seconds...") %
+                         self.conf.kombu_reconnect_delay)
+                time.sleep(self.conf.kombu_reconnect_delay)
+
+            try:
+                self.connection.release()
+            except self.connection_errors:
+                pass
+            self.connection = None
+
+    def reconnect(self, retry=None):
+        """Handles reconnecting and re-establishing queues.
+        Will retry up to retry number of times.
+        retry = None means use the value of rabbit_max_retries
+        retry = -1 means to retry forever
+        retry = 0 means no retry
+        retry = N means N retries
+        Sleep between tries, starting at self.interval_start
+        seconds, backing off self.interval_stepping number of seconds
+        each attempt.
+        """
+
+        attempt = 0
+        loop_forever = False
+        if retry is None:
+            retry = self.max_retries
+        if retry is None or retry < 0:
+            loop_forever = True
+
+        while True:
+            self._disconnect()
+
+            broker = six.next(self.brokers)
+            attempt += 1
+            try:
+                self._connect(broker)
+                return
+            except IOError as ex:
+                e = ex
+            except self.connection_errors as ex:
+                e = ex
+            except Exception as ex:
+                # NOTE(comstud): Unfortunately it's possible for amqplib
+                # to return an error not covered by its transport
+                # connection_errors in the case of a timeout waiting for
+                # a protocol response.  (See paste link in LP888621)
+                # So, we check all exceptions for 'timeout' in them
+                # and try to reconnect in this case.
+                if 'timeout' not in six.text_type(e):
+                    raise
+                e = ex
+
+            log_info = {}
+            log_info['err_str'] = e
+            log_info['retry'] = retry or 0
+            log_info.update(broker)
+
+            if not loop_forever and attempt > retry:
+                msg = _('Unable to connect to AMQP server on '
+                        '%(hostname)s:%(port)d after %(retry)d '
+                        'tries: %(err_str)s') % log_info
+                LOG.error(msg)
+                raise exceptions.MessageDeliveryFailure(msg)
+            else:
+                if attempt == 1:
+                    sleep_time = self.interval_start or 1
+                elif attempt > 1:
+                    sleep_time += self.interval_stepping
+
+                sleep_time = min(sleep_time, self.interval_max)
+
+                log_info['sleep_time'] = sleep_time
+                if 'Socket closed' in six.text_type(e):
+                    LOG.error(_('AMQP server %(hostname)s:%(port)d closed'
+                                ' the connection. Check login credentials:'
+                                ' %(err_str)s'), log_info)
+                else:
+                    LOG.error(_('AMQP server on %(hostname)s:%(port)d is '
+                                'unreachable: %(err_str)s. Trying again in '
+                                '%(sleep_time)d seconds.'), log_info)
+                time.sleep(sleep_time)
+
+    def ensure(self, error_callback, method, retry=None):
+        while True:
+            try:
+                return method()
+            except self.connection_errors as e:
+                if error_callback:
+                    error_callback(e)
+            except self.channel_errors as e:
+                if error_callback:
+                    error_callback(e)
+            except (socket.timeout, IOError) as e:
+                if error_callback:
+                    error_callback(e)
+            except Exception as e:
+                # NOTE(comstud): Unfortunately it's possible for amqplib
+                # to return an error not covered by its transport
+                # connection_errors in the case of a timeout waiting for
+                # a protocol response.  (See paste link in LP888621)
+                # So, we check all exceptions for 'timeout' in them
+                # and try to reconnect in this case.
+                if 'timeout' not in six.text_type(e):
+                    raise
+                if error_callback:
+                    error_callback(e)
+            self.reconnect(retry=retry)
+
+    def get_channel(self):
+        """Convenience call for bin/clear_rabbit_queues."""
+        return self.channel
+
+    def close(self):
+        """Close/release this connection."""
+        if self.connection:
+            self.connection.release()
+            self.connection = None
+
+    def reset(self):
+        """Reset a connection so it can be used again."""
+        self.channel.close()
+        self.channel = self.connection.channel()
+        # work around 'memory' transport bug in 1.1.3
+        if self.memory_transport:
+            self.channel._new_queue('ae.undeliver')
+        self.consumers = []
+
+    def declare_consumer(self, consumer_cls, topic, callback):
+        """Create a Consumer using the class that was passed in and
+        add it to our list of consumers
+        """
+
+        def _connect_error(exc):
+            log_info = {'topic': topic, 'err_str': exc}
+            LOG.error(_("Failed to declare consumer for topic '%(topic)s': "
+                      "%(err_str)s"), log_info)
+
+        def _declare_consumer():
+            consumer = consumer_cls(self.conf, self.channel, topic, callback,
+                                    six.next(self.consumer_num))
+            self.consumers.append(consumer)
+            return consumer
+
+        return self.ensure(_connect_error, _declare_consumer)
+
+    def iterconsume(self, limit=None, timeout=None):
+        """Return an iterator that will consume from all queues/consumers."""
+
+        def _error_callback(exc):
+            if isinstance(exc, socket.timeout):
+                LOG.debug('Timed out waiting for RPC response: %s', exc)
+                raise rpc_common.Timeout()
+            else:
+                LOG.exception(_('Failed to consume message from queue: %s'),
+                              exc)
+                self.do_consume = True
+
+        def _consume():
+            if self.do_consume:
+                queues_head = self.consumers[:-1]  # not fanout.
+                queues_tail = self.consumers[-1]  # fanout
+                for queue in queues_head:
+                    queue.consume(nowait=True)
+                queues_tail.consume(nowait=False)
+                self.do_consume = False
+            return self.connection.drain_events(timeout=timeout)
+
+        for iteration in itertools.count(0):
+            if limit and iteration >= limit:
+                raise StopIteration
+            yield self.ensure(_error_callback, _consume)
+
+    def publisher_send(self, cls, topic, msg, timeout=None, retry=None,
+                       **kwargs):
+        """Send to a publisher based on the publisher class."""
+
+        def _error_callback(exc):
+            log_info = {'topic': topic, 'err_str': exc}
+            LOG.exception(_("Failed to publish message to topic "
+                          "'%(topic)s': %(err_str)s"), log_info)
+
+        def _publish():
+            publisher = cls(self.conf, self.channel, topic=topic, **kwargs)
+            publisher.send(msg, timeout)
+
+        self.ensure(_error_callback, _publish, retry=retry)
+
+    def declare_direct_consumer(self, topic, callback):
+        """Create a 'direct' queue.
+        In nova's use, this is generally a msg_id queue used for
+        responses for call/multicall
+        """
+        self.declare_consumer(DirectConsumer, topic, callback)
+
+    def declare_topic_consumer(self, exchange_name, topic, callback=None,
+                               queue_name=None):
+        """Create a 'topic' consumer."""
+        self.declare_consumer(functools.partial(TopicConsumer,
+                                                name=queue_name,
+                                                exchange_name=exchange_name,
+                                                ),
+                              topic, callback)
+
+    def declare_fanout_consumer(self, topic, callback):
+        """Create a 'fanout' consumer."""
+        self.declare_consumer(FanoutConsumer, topic, callback)
+
+    def direct_send(self, msg_id, msg):
+        """Send a 'direct' message."""
+        self.publisher_send(DirectPublisher, msg_id, msg)
+
+    def topic_send(self, exchange_name, topic, msg, timeout=None, retry=None):
+        """Send a 'topic' message."""
+        self.publisher_send(TopicPublisher, topic, msg, timeout,
+                            exchange_name=exchange_name, retry=retry)
+
+    def fanout_send(self, topic, msg, retry=None):
+        """Send a 'fanout' message."""
+        self.publisher_send(FanoutPublisher, topic, msg, retry=retry)
+
+    def notify_send(self, exchange_name, topic, msg, retry=None, **kwargs):
+        """Send a notify message on a topic."""
+        self.publisher_send(NotifyPublisher, topic, msg, timeout=None,
+                            exchange_name=exchange_name, retry=retry, **kwargs)
+
+    def consume(self, limit=None, timeout=None):
+        """Consume from all queues/consumers."""
+        it = self.iterconsume(limit=limit, timeout=timeout)
+        while True:
+            try:
+                six.next(it)
+            except StopIteration:
+                return
+
+
+class RabbitDriver(amqpdriver.AMQPDriverBase):
+
+    def __init__(self, conf, url,
+                 default_exchange=None,
+                 allowed_remote_exmods=None):
+        conf.register_opts(rabbit_opts)
+        conf.register_opts(rpc_amqp.amqp_opts)
+
+        connection_pool = rpc_amqp.get_connection_pool(conf, url, Connection)
+
+        super(RabbitDriver, self).__init__(conf, url,
+                                           connection_pool,
+                                           default_exchange,
+                                           allowed_remote_exmods)
+
+    def require_features(self, requeue=True):
+        pass
 
=== added directory '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch'
=== added file '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/.timestamp'
=== added directory '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/oslo'
=== added directory '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/oslo/messaging'
=== added directory '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/oslo/messaging/_drivers'
=== added file '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/oslo/messaging/_drivers/common.py'
--- .pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/oslo/messaging/_drivers/common.py	1970-01-01 00:00:00 +0000
+++ .pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/oslo/messaging/_drivers/common.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,374 @@
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+# Copyright 2011 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import copy
+import logging
+import sys
+import time
+import traceback
+
+import six
+
+from oslo import messaging
+from oslo.messaging import _utils as utils
+from oslo.messaging.openstack.common.gettextutils import _
+from oslo.messaging.openstack.common import jsonutils
+
+LOG = logging.getLogger(__name__)
+
+_EXCEPTIONS_MODULE = 'exceptions' if six.PY2 else 'builtins'
+
+
+'''RPC Envelope Version.
+
+This version number applies to the top level structure of messages sent out.
+It does *not* apply to the message payload, which must be versioned
+independently.  For example, when using rpc APIs, a version number is applied
+for changes to the API being exposed over rpc.  This version number is handled
+in the rpc proxy and dispatcher modules.
+
+This version number applies to the message envelope that is used in the
+serialization done inside the rpc layer.  See serialize_msg() and
+deserialize_msg().
+
+The current message format (version 2.0) is very simple.  It is:
+
+    {
+        'oslo.version': <RPC Envelope Version as a String>,
+        'oslo.message': <Application Message Payload, JSON encoded>
+    }
+
+Message format version '1.0' is just considered to be the messages we sent
+without a message envelope.
+
+So, the current message envelope just includes the envelope version.  It may
+eventually contain additional information, such as a signature for the message
+payload.
+
+We will JSON encode the application message payload.  The message envelope,
+which includes the JSON encoded application message body, will be passed down
+to the messaging libraries as a dict.
+'''
+_RPC_ENVELOPE_VERSION = '2.0'
+
+_VERSION_KEY = 'oslo.version'
+_MESSAGE_KEY = 'oslo.message'
+
+_REMOTE_POSTFIX = '_Remote'
+
+
+class RPCException(Exception):
+    msg_fmt = _("An unknown RPC related exception occurred.")
+
+    def __init__(self, message=None, **kwargs):
+        self.kwargs = kwargs
+
+        if not message:
+            try:
+                message = self.msg_fmt % kwargs
+
+            except Exception:
+                # kwargs doesn't match a variable in the message
+                # log the issue and the kwargs
+                LOG.exception(_('Exception in string format operation'))
+                for name, value in six.iteritems(kwargs):
+                    LOG.error("%s: %s", name, value)
+                # at least get the core message out if something happened
+                message = self.msg_fmt
+
+        super(RPCException, self).__init__(message)
+
+
+class Timeout(RPCException):
+    """Signifies that a timeout has occurred.
+
+    This exception is raised if the rpc_response_timeout is reached while
+    waiting for a response from the remote side.
+    """
+    msg_fmt = _('Timeout while waiting on RPC response - '
+                'topic: "%(topic)s", RPC method: "%(method)s" '
+                'info: "%(info)s"')
+
+    def __init__(self, info=None, topic=None, method=None):
+        """Initiates Timeout object.
+
+        :param info: Extra info to convey to the user
+        :param topic: The topic that the rpc call was sent to
+        :param rpc_method_name: The name of the rpc method being
+                                called
+        """
+        self.info = info
+        self.topic = topic
+        self.method = method
+        super(Timeout, self).__init__(
+            None,
+            info=info or _('<unknown>'),
+            topic=topic or _('<unknown>'),
+            method=method or _('<unknown>'))
+
+
+class DuplicateMessageError(RPCException):
+    msg_fmt = _("Found duplicate message(%(msg_id)s). Skipping it.")
+
+
+class InvalidRPCConnectionReuse(RPCException):
+    msg_fmt = _("Invalid reuse of an RPC connection.")
+
+
+class UnsupportedRpcVersion(RPCException):
+    msg_fmt = _("Specified RPC version, %(version)s, not supported by "
+                "this endpoint.")
+
+
+class UnsupportedRpcEnvelopeVersion(RPCException):
+    msg_fmt = _("Specified RPC envelope version, %(version)s, "
+                "not supported by this endpoint.")
+
+
+class RpcVersionCapError(RPCException):
+    msg_fmt = _("Specified RPC version cap, %(version_cap)s, is too low")
+
+
+class Connection(object):
+    """A connection, returned by rpc.create_connection().
+
+    This class represents a connection to the message bus used for rpc.
+    An instance of this class should never be created by users of the rpc API.
+    Use rpc.create_connection() instead.
+    """
+    def close(self):
+        """Close the connection.
+
+        This method must be called when the connection will no longer be used.
+        It will ensure that any resources associated with the connection, such
+        as a network connection, and cleaned up.
+        """
+        raise NotImplementedError()
+
+
+def _safe_log(log_func, msg, msg_data):
+    """Sanitizes the msg_data field before logging."""
+    SANITIZE = ['_context_auth_token', 'auth_token', 'new_pass']
+
+    def _fix_passwords(d):
+        """Sanitizes the password fields in the dictionary."""
+        for k in six.iterkeys(d):
+            if k.lower().find('password') != -1:
+                d[k] = '<SANITIZED>'
+            elif k.lower() in SANITIZE:
+                d[k] = '<SANITIZED>'
+            elif isinstance(d[k], dict):
+                _fix_passwords(d[k])
+        return d
+
+    return log_func(msg, _fix_passwords(copy.deepcopy(msg_data)))
+
+
+def serialize_remote_exception(failure_info, log_failure=True):
+    """Prepares exception data to be sent over rpc.
+
+    Failure_info should be a sys.exc_info() tuple.
+
+    """
+    tb = traceback.format_exception(*failure_info)
+    failure = failure_info[1]
+    if log_failure:
+        LOG.error(_("Returning exception %s to caller"),
+                  six.text_type(failure))
+        LOG.error(tb)
+
+    kwargs = {}
+    if hasattr(failure, 'kwargs'):
+        kwargs = failure.kwargs
+
+    # NOTE(matiu): With cells, it's possible to re-raise remote, remote
+    # exceptions. Lets turn it back into the original exception type.
+    cls_name = six.text_type(failure.__class__.__name__)
+    mod_name = six.text_type(failure.__class__.__module__)
+    if (cls_name.endswith(_REMOTE_POSTFIX) and
+            mod_name.endswith(_REMOTE_POSTFIX)):
+        cls_name = cls_name[:-len(_REMOTE_POSTFIX)]
+        mod_name = mod_name[:-len(_REMOTE_POSTFIX)]
+
+    data = {
+        'class': cls_name,
+        'module': mod_name,
+        'message': six.text_type(failure),
+        'tb': tb,
+        'args': failure.args,
+        'kwargs': kwargs
+    }
+
+    json_data = jsonutils.dumps(data)
+
+    return json_data
+
+
+def deserialize_remote_exception(data, allowed_remote_exmods):
+    failure = jsonutils.loads(six.text_type(data))
+
+    trace = failure.get('tb', [])
+    message = failure.get('message', "") + "\n" + "\n".join(trace)
+    name = failure.get('class')
+    module = failure.get('module')
+
+    # NOTE(ameade): We DO NOT want to allow just any module to be imported, in
+    # order to prevent arbitrary code execution.
+    if module != _EXCEPTIONS_MODULE and module not in allowed_remote_exmods:
+        return messaging.RemoteError(name, failure.get('message'), trace)
+
+    try:
+        __import__(module)
+        mod = sys.modules[module]
+        klass = getattr(mod, name)
+        if not issubclass(klass, Exception):
+            raise TypeError("Can only deserialize Exceptions")
+
+        failure = klass(*failure.get('args', []), **failure.get('kwargs', {}))
+    except (AttributeError, TypeError, ImportError):
+        return messaging.RemoteError(name, failure.get('message'), trace)
+
+    ex_type = type(failure)
+    str_override = lambda self: message
+    new_ex_type = type(ex_type.__name__ + _REMOTE_POSTFIX, (ex_type,),
+                       {'__str__': str_override, '__unicode__': str_override})
+    new_ex_type.__module__ = '%s%s' % (module, _REMOTE_POSTFIX)
+    try:
+        # NOTE(ameade): Dynamically create a new exception type and swap it in
+        # as the new type for the exception. This only works on user defined
+        # Exceptions and not core Python exceptions. This is important because
+        # we cannot necessarily change an exception message so we must override
+        # the __str__ method.
+        failure.__class__ = new_ex_type
+    except TypeError:
+        # NOTE(ameade): If a core exception then just add the traceback to the
+        # first exception argument.
+        failure.args = (message,) + failure.args[1:]
+    return failure
+
+
+class CommonRpcContext(object):
+    def __init__(self, **kwargs):
+        self.values = kwargs
+
+    def __getattr__(self, key):
+        try:
+            return self.values[key]
+        except KeyError:
+            raise AttributeError(key)
+
+    def to_dict(self):
+        return copy.deepcopy(self.values)
+
+    @classmethod
+    def from_dict(cls, values):
+        return cls(**values)
+
+    def deepcopy(self):
+        return self.from_dict(self.to_dict())
+
+    def update_store(self):
+        # local.store.context = self
+        pass
+
+
+class ClientException(Exception):
+    """Encapsulates actual exception expected to be hit by a RPC proxy object.
+
+    Merely instantiating it records the current exception information, which
+    will be passed back to the RPC client without exceptional logging.
+    """
+    def __init__(self):
+        self._exc_info = sys.exc_info()
+
+
+def serialize_msg(raw_msg):
+    # NOTE(russellb) See the docstring for _RPC_ENVELOPE_VERSION for more
+    # information about this format.
+    msg = {_VERSION_KEY: _RPC_ENVELOPE_VERSION,
+           _MESSAGE_KEY: jsonutils.dumps(raw_msg)}
+
+    return msg
+
+
+def deserialize_msg(msg):
+    # NOTE(russellb): Hang on to your hats, this road is about to
+    # get a little bumpy.
+    #
+    # Robustness Principle:
+    #    "Be strict in what you send, liberal in what you accept."
+    #
+    # At this point we have to do a bit of guessing about what it
+    # is we just received.  Here is the set of possibilities:
+    #
+    # 1) We received a dict.  This could be 2 things:
+    #
+    #   a) Inspect it to see if it looks like a standard message envelope.
+    #      If so, great!
+    #
+    #   b) If it doesn't look like a standard message envelope, it could either
+    #      be a notification, or a message from before we added a message
+    #      envelope (referred to as version 1.0).
+    #      Just return the message as-is.
+    #
+    # 2) It's any other non-dict type.  Just return it and hope for the best.
+    #    This case covers return values from rpc.call() from before message
+    #    envelopes were used.  (messages to call a method were always a dict)
+
+    if not isinstance(msg, dict):
+        # See #2 above.
+        return msg
+
+    base_envelope_keys = (_VERSION_KEY, _MESSAGE_KEY)
+    if not all(map(lambda key: key in msg, base_envelope_keys)):
+        #  See #1.b above.
+        return msg
+
+    # At this point we think we have the message envelope
+    # format we were expecting. (#1.a above)
+
+    if not utils.version_is_compatible(_RPC_ENVELOPE_VERSION,
+                                       msg[_VERSION_KEY]):
+        raise UnsupportedRpcEnvelopeVersion(version=msg[_VERSION_KEY])
+
+    raw_msg = jsonutils.loads(msg[_MESSAGE_KEY])
+
+    return raw_msg
+
+
+class DecayingTimer(object):
+    def __init__(self, duration=None):
+        self._duration = duration
+        self._ends_at = None
+
+    def start(self):
+        if self._duration is not None:
+            self._ends_at = time.time() + max(0, self._duration)
+
+    def check_return(self, timeout_callback, *args, **kwargs):
+        if self._duration is None:
+            return None
+        if self._ends_at is None:
+            raise RuntimeError(_("Can not check/return a timeout from a timer"
+                               " that has not been started."))
+
+        maximum = kwargs.pop('maximum', None)
+        left = self._ends_at - time.time()
+        if left <= 0:
+            timeout_callback(*args, **kwargs)
+
+        return left if maximum is None else min(left, maximum)
 
=== added directory '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/tests'
=== added file '.pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/tests/test_utils.py'
--- .pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/tests/test_utils.py	1970-01-01 00:00:00 +0000
+++ .pc/0002-Fix-_poll_connection-not-timeout-issue-2-2.patch/tests/test_utils.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,49 @@
+
+# Copyright 2013 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from oslo.messaging import _utils as utils
+from tests import utils as test_utils
+
+
+class VersionIsCompatibleTestCase(test_utils.BaseTestCase):
+    def test_version_is_compatible_same(self):
+        self.assertTrue(utils.version_is_compatible('1.23', '1.23'))
+
+    def test_version_is_compatible_newer_minor(self):
+        self.assertTrue(utils.version_is_compatible('1.24', '1.23'))
+
+    def test_version_is_compatible_older_minor(self):
+        self.assertFalse(utils.version_is_compatible('1.22', '1.23'))
+
+    def test_version_is_compatible_major_difference1(self):
+        self.assertFalse(utils.version_is_compatible('2.23', '1.23'))
+
+    def test_version_is_compatible_major_difference2(self):
+        self.assertFalse(utils.version_is_compatible('1.23', '2.23'))
+
+    def test_version_is_compatible_newer_rev(self):
+        self.assertFalse(utils.version_is_compatible('1.23', '1.23.1'))
+
+    def test_version_is_compatible_newer_rev_both(self):
+        self.assertFalse(utils.version_is_compatible('1.23.1', '1.23.2'))
+
+    def test_version_is_compatible_older_rev_both(self):
+        self.assertTrue(utils.version_is_compatible('1.23.2', '1.23.1'))
+
+    def test_version_is_compatible_older_rev(self):
+        self.assertTrue(utils.version_is_compatible('1.24', '1.23.1'))
+
+    def test_version_is_compatible_no_rev_is_zero(self):
+        self.assertTrue(utils.version_is_compatible('1.23.0', '1.23'))
 
=== added directory '.pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch'
=== added file '.pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/.timestamp'
=== added directory '.pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/oslo'
=== added directory '.pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/oslo/messaging'
=== added directory '.pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/oslo/messaging/_drivers'
=== added file '.pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/oslo/messaging/_drivers/impl_qpid.py'
--- .pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/oslo/messaging/_drivers/impl_qpid.py	1970-01-01 00:00:00 +0000
+++ .pc/0003-qpid-Always-auto-delete-queue-of-DirectConsumer.patch/oslo/messaging/_drivers/impl_qpid.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,729 @@
+#    Copyright 2011 OpenStack Foundation
+#    Copyright 2011 - 2012, Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+import itertools
+import logging
+import random
+import time
+
+import six
+
+from oslo.config import cfg
+from oslo.messaging._drivers import amqp as rpc_amqp
+from oslo.messaging._drivers import amqpdriver
+from oslo.messaging._drivers import common as rpc_common
+from oslo.messaging import exceptions
+from oslo.messaging.openstack.common.gettextutils import _
+from oslo.messaging.openstack.common import jsonutils
+from oslo.utils import importutils
+from oslo.utils import netutils
+
+qpid_codec = importutils.try_import("qpid.codec010")
+qpid_messaging = importutils.try_import("qpid.messaging")
+qpid_exceptions = importutils.try_import("qpid.messaging.exceptions")
+
+LOG = logging.getLogger(__name__)
+
+qpid_opts = [
+    cfg.StrOpt('qpid_hostname',
+               default='localhost',
+               help='Qpid broker hostname.'),
+    cfg.IntOpt('qpid_port',
+               default=5672,
+               help='Qpid broker port.'),
+    cfg.ListOpt('qpid_hosts',
+                default=['$qpid_hostname:$qpid_port'],
+                help='Qpid HA cluster host:port pairs.'),
+    cfg.StrOpt('qpid_username',
+               default='',
+               help='Username for Qpid connection.'),
+    cfg.StrOpt('qpid_password',
+               default='',
+               help='Password for Qpid connection.',
+               secret=True),
+    cfg.StrOpt('qpid_sasl_mechanisms',
+               default='',
+               help='Space separated list of SASL mechanisms to use for '
+                    'auth.'),
+    cfg.IntOpt('qpid_heartbeat',
+               default=60,
+               help='Seconds between connection keepalive heartbeats.'),
+    cfg.StrOpt('qpid_protocol',
+               default='tcp',
+               help="Transport to use, either 'tcp' or 'ssl'."),
+    cfg.BoolOpt('qpid_tcp_nodelay',
+                default=True,
+                help='Whether to disable the Nagle algorithm.'),
+    cfg.IntOpt('qpid_receiver_capacity',
+               default=1,
+               help='The number of prefetched messages held by receiver.'),
+    # NOTE(russellb) If any additional versions are added (beyond 1 and 2),
+    # this file could probably use some additional refactoring so that the
+    # differences between each version are split into different classes.
+    cfg.IntOpt('qpid_topology_version',
+               default=1,
+               help="The qpid topology version to use.  Version 1 is what "
+                    "was originally used by impl_qpid.  Version 2 includes "
+                    "some backwards-incompatible changes that allow broker "
+                    "federation to work.  Users should update to version 2 "
+                    "when they are able to take everything down, as it "
+                    "requires a clean break."),
+]
+
+JSON_CONTENT_TYPE = 'application/json; charset=utf8'
+
+
+def raise_invalid_topology_version(conf):
+    msg = (_("Invalid value for qpid_topology_version: %d") %
+           conf.qpid_topology_version)
+    LOG.error(msg)
+    raise Exception(msg)
+
+
+class QpidMessage(dict):
+    def __init__(self, session, raw_message):
+        super(QpidMessage, self).__init__(
+            rpc_common.deserialize_msg(raw_message.content))
+        self._raw_message = raw_message
+        self._session = session
+
+    def acknowledge(self):
+        self._session.acknowledge(self._raw_message)
+
+    def requeue(self):
+        pass
+
+
+class ConsumerBase(object):
+    """Consumer base class."""
+
+    def __init__(self, conf, session, callback, node_name, node_opts,
+                 link_name, link_opts):
+        """Declare a queue on an amqp session.
+
+        'session' is the amqp session to use
+        'callback' is the callback to call when messages are received
+        'node_name' is the first part of the Qpid address string, before ';'
+        'node_opts' will be applied to the "x-declare" section of "node"
+                    in the address string.
+        'link_name' goes into the "name" field of the "link" in the address
+                    string
+        'link_opts' will be applied to the "x-declare" section of "link"
+                    in the address string.
+        """
+        self.callback = callback
+        self.receiver = None
+        self.rcv_capacity = conf.qpid_receiver_capacity
+        self.session = None
+
+        if conf.qpid_topology_version == 1:
+            addr_opts = {
+                "create": "always",
+                "node": {
+                    "type": "topic",
+                    "x-declare": {
+                        "durable": True,
+                        "auto-delete": True,
+                    },
+                },
+                "link": {
+                    "durable": True,
+                    "x-declare": {
+                        "durable": False,
+                        "auto-delete": True,
+                        "exclusive": False,
+                    },
+                },
+            }
+            addr_opts["node"]["x-declare"].update(node_opts)
+        elif conf.qpid_topology_version == 2:
+            addr_opts = {
+                "link": {
+                    "x-declare": {
+                        "auto-delete": True,
+                        "exclusive": False,
+                    },
+                },
+            }
+        else:
+            raise_invalid_topology_version(conf)
+
+        addr_opts["link"]["x-declare"].update(link_opts)
+        if link_name:
+            addr_opts["link"]["name"] = link_name
+
+        self.address = "%s ; %s" % (node_name, jsonutils.dumps(addr_opts))
+
+        self.connect(session)
+
+    def connect(self, session):
+        """Declare the receiver on connect."""
+        self._declare_receiver(session)
+
+    def reconnect(self, session):
+        """Re-declare the receiver after a Qpid reconnect."""
+        self._declare_receiver(session)
+
+    def _declare_receiver(self, session):
+        self.session = session
+        self.receiver = session.receiver(self.address)
+        self.receiver.capacity = self.rcv_capacity
+
+    def _unpack_json_msg(self, msg):
+        """Load the JSON data in msg if msg.content_type indicates that it
+           is necessary.  Put the loaded data back into msg.content and
+           update msg.content_type appropriately.
+
+        A Qpid Message containing a dict will have a content_type of
+        'amqp/map', whereas one containing a string that needs to be converted
+        back from JSON will have a content_type of JSON_CONTENT_TYPE.
+
+        :param msg: a Qpid Message object
+        :returns: None
+        """
+        if msg.content_type == JSON_CONTENT_TYPE:
+            msg.content = jsonutils.loads(msg.content)
+            msg.content_type = 'amqp/map'
+
+    def consume(self):
+        """Fetch the message and pass it to the callback object."""
+        message = self.receiver.fetch()
+        try:
+            self._unpack_json_msg(message)
+            self.callback(QpidMessage(self.session, message))
+        except Exception:
+            LOG.exception(_("Failed to process message... skipping it."))
+            self.session.acknowledge(message)
+
+    def get_receiver(self):
+        return self.receiver
+
+    def get_node_name(self):
+        return self.address.split(';')[0]
+
+
+class DirectConsumer(ConsumerBase):
+    """Queue/consumer class for 'direct'."""
+
+    def __init__(self, conf, session, msg_id, callback):
+        """Init a 'direct' queue.
+
+        'session' is the amqp session to use
+        'msg_id' is the msg_id to listen on
+        'callback' is the callback to call when messages are received
+        """
+
+        link_opts = {
+            "auto-delete": conf.amqp_auto_delete,
+            "exclusive": True,
+            "durable": conf.amqp_durable_queues,
+        }
+
+        if conf.qpid_topology_version == 1:
+            node_name = "%s/%s" % (msg_id, msg_id)
+            node_opts = {"type": "direct"}
+            link_name = msg_id
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.direct/%s" % msg_id
+            node_opts = {}
+            link_name = msg_id
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(DirectConsumer, self).__init__(conf, session, callback,
+                                             node_name, node_opts, link_name,
+                                             link_opts)
+
+
+class TopicConsumer(ConsumerBase):
+    """Consumer class for 'topic'."""
+
+    def __init__(self, conf, session, topic, callback, exchange_name,
+                 name=None):
+        """Init a 'topic' queue.
+
+        :param session: the amqp session to use
+        :param topic: is the topic to listen on
+        :paramtype topic: str
+        :param callback: the callback to call when messages are received
+        :param name: optional queue name, defaults to topic
+        """
+
+        link_opts = {
+            "auto-delete": conf.amqp_auto_delete,
+            "durable": conf.amqp_durable_queues,
+        }
+
+        if conf.qpid_topology_version == 1:
+            node_name = "%s/%s" % (exchange_name, topic)
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.topic/topic/%s/%s" % (exchange_name, topic)
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(TopicConsumer, self).__init__(conf, session, callback, node_name,
+                                            {}, name or topic, link_opts)
+
+
+class FanoutConsumer(ConsumerBase):
+    """Consumer class for 'fanout'."""
+
+    def __init__(self, conf, session, topic, callback):
+        """Init a 'fanout' queue.
+
+        'session' is the amqp session to use
+        'topic' is the topic to listen on
+        'callback' is the callback to call when messages are received
+        """
+        self.conf = conf
+
+        link_opts = {"exclusive": True}
+
+        if conf.qpid_topology_version == 1:
+            node_name = "%s_fanout" % topic
+            node_opts = {"durable": False, "type": "fanout"}
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.topic/fanout/%s" % topic
+            node_opts = {}
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(FanoutConsumer, self).__init__(conf, session, callback,
+                                             node_name, node_opts, None,
+                                             link_opts)
+
+
+class Publisher(object):
+    """Base Publisher class."""
+
+    def __init__(self, conf, session, node_name, node_opts=None):
+        """Init the Publisher class with the exchange_name, routing_key,
+        and other options
+        """
+        self.sender = None
+        self.session = session
+
+        if conf.qpid_topology_version == 1:
+            addr_opts = {
+                "create": "always",
+                "node": {
+                    "type": "topic",
+                    "x-declare": {
+                        "durable": False,
+                        # auto-delete isn't implemented for exchanges in qpid,
+                        # but put in here anyway
+                        "auto-delete": True,
+                    },
+                },
+            }
+            if node_opts:
+                addr_opts["node"]["x-declare"].update(node_opts)
+
+            self.address = "%s ; %s" % (node_name, jsonutils.dumps(addr_opts))
+        elif conf.qpid_topology_version == 2:
+            self.address = node_name
+        else:
+            raise_invalid_topology_version(conf)
+
+        self.reconnect(session)
+
+    def reconnect(self, session):
+        """Re-establish the Sender after a reconnection."""
+        self.sender = session.sender(self.address)
+
+    def _pack_json_msg(self, msg):
+        """Qpid cannot serialize dicts containing strings longer than 65535
+           characters.  This function dumps the message content to a JSON
+           string, which Qpid is able to handle.
+
+        :param msg: May be either a Qpid Message object or a bare dict.
+        :returns: A Qpid Message with its content field JSON encoded.
+        """
+        try:
+            msg.content = jsonutils.dumps(msg.content)
+        except AttributeError:
+            # Need to have a Qpid message so we can set the content_type.
+            msg = qpid_messaging.Message(jsonutils.dumps(msg))
+        msg.content_type = JSON_CONTENT_TYPE
+        return msg
+
+    def send(self, msg):
+        """Send a message."""
+        try:
+            # Check if Qpid can encode the message
+            check_msg = msg
+            if not hasattr(check_msg, 'content_type'):
+                check_msg = qpid_messaging.Message(msg)
+            content_type = check_msg.content_type
+            enc, dec = qpid_messaging.message.get_codec(content_type)
+            enc(check_msg.content)
+        except qpid_codec.CodecException:
+            # This means the message couldn't be serialized as a dict.
+            msg = self._pack_json_msg(msg)
+        self.sender.send(msg)
+
+
+class DirectPublisher(Publisher):
+    """Publisher class for 'direct'."""
+    def __init__(self, conf, session, topic):
+        """Init a 'direct' publisher."""
+
+        if conf.qpid_topology_version == 1:
+            node_name = "%s/%s" % (topic, topic)
+            node_opts = {"type": "direct"}
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.direct/%s" % topic
+            node_opts = {}
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(DirectPublisher, self).__init__(conf, session, node_name,
+                                              node_opts)
+
+
+class TopicPublisher(Publisher):
+    """Publisher class for 'topic'."""
+    def __init__(self, conf, session, exchange_name, topic):
+        """Init a 'topic' publisher.
+        """
+        if conf.qpid_topology_version == 1:
+            node_name = "%s/%s" % (exchange_name, topic)
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.topic/topic/%s/%s" % (exchange_name, topic)
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(TopicPublisher, self).__init__(conf, session, node_name)
+
+
+class FanoutPublisher(Publisher):
+    """Publisher class for 'fanout'."""
+    def __init__(self, conf, session, topic):
+        """Init a 'fanout' publisher.
+        """
+
+        if conf.qpid_topology_version == 1:
+            node_name = "%s_fanout" % topic
+            node_opts = {"type": "fanout"}
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.topic/fanout/%s" % topic
+            node_opts = {}
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(FanoutPublisher, self).__init__(conf, session, node_name,
+                                              node_opts)
+
+
+class NotifyPublisher(Publisher):
+    """Publisher class for notifications."""
+    def __init__(self, conf, session, exchange_name, topic):
+        """Init a 'topic' publisher.
+        """
+        node_opts = {"durable": True}
+
+        if conf.qpid_topology_version == 1:
+            node_name = "%s/%s" % (exchange_name, topic)
+        elif conf.qpid_topology_version == 2:
+            node_name = "amq.topic/topic/%s/%s" % (exchange_name, topic)
+        else:
+            raise_invalid_topology_version(conf)
+
+        super(NotifyPublisher, self).__init__(conf, session, node_name,
+                                              node_opts)
+
+
+class Connection(object):
+    """Connection object."""
+
+    pools = {}
+
+    def __init__(self, conf, url):
+        if not qpid_messaging:
+            raise ImportError("Failed to import qpid.messaging")
+
+        self.connection = None
+        self.session = None
+        self.consumers = {}
+        self.conf = conf
+
+        self.brokers_params = []
+        if url.hosts:
+            for host in url.hosts:
+                params = {
+                    'username': host.username or '',
+                    'password': host.password or '',
+                }
+                if host.port is not None:
+                    params['host'] = '%s:%d' % (host.hostname, host.port)
+                else:
+                    params['host'] = host.hostname
+                self.brokers_params.append(params)
+        else:
+            # Old configuration format
+            for adr in self.conf.qpid_hosts:
+                hostname, port = netutils.parse_host_port(
+                    adr, default_port=5672)
+
+                params = {
+                    'host': '%s:%d' % (hostname, port),
+                    'username': self.conf.qpid_username,
+                    'password': self.conf.qpid_password,
+                }
+                self.brokers_params.append(params)
+
+        random.shuffle(self.brokers_params)
+        self.brokers = itertools.cycle(self.brokers_params)
+
+        self.reconnect()
+
+    def _connect(self, broker):
+        # Create the connection - this does not open the connection
+        self.connection = qpid_messaging.Connection(broker['host'])
+
+        # Check if flags are set and if so set them for the connection
+        # before we call open
+        self.connection.username = broker['username']
+        self.connection.password = broker['password']
+
+        self.connection.sasl_mechanisms = self.conf.qpid_sasl_mechanisms
+        # Reconnection is done by self.reconnect()
+        self.connection.reconnect = False
+        self.connection.heartbeat = self.conf.qpid_heartbeat
+        self.connection.transport = self.conf.qpid_protocol
+        self.connection.tcp_nodelay = self.conf.qpid_tcp_nodelay
+        self.connection.open()
+
+    def _register_consumer(self, consumer):
+        self.consumers[six.text_type(consumer.get_receiver())] = consumer
+
+    def _lookup_consumer(self, receiver):
+        return self.consumers[six.text_type(receiver)]
+
+    def _disconnect(self):
+        # Close the session if necessary
+        if self.connection is not None and self.connection.opened():
+            try:
+                self.connection.close()
+            except qpid_exceptions.MessagingError:
+                pass
+        self.connection = None
+
+    def reconnect(self, retry=None):
+        """Handles reconnecting and re-establishing sessions and queues.
+        Will retry up to retry number of times.
+        retry = None or -1 means to retry forever
+        retry = 0 means no retry
+        retry = N means N retries
+        """
+        delay = 1
+        attempt = 0
+        loop_forever = False
+        if retry is None or retry < 0:
+            loop_forever = True
+
+        while True:
+            self._disconnect()
+
+            attempt += 1
+            broker = six.next(self.brokers)
+            try:
+                self._connect(broker)
+            except qpid_exceptions.MessagingError as e:
+                msg_dict = dict(e=e,
+                                delay=delay,
+                                retry=retry,
+                                broker=broker)
+                if not loop_forever and attempt > retry:
+                    msg = _('Unable to connect to AMQP server on '
+                            '%(broker)s after %(retry)d '
+                            'tries: %(e)s') % msg_dict
+                    LOG.error(msg)
+                    raise exceptions.MessageDeliveryFailure(msg)
+                else:
+                    msg = _("Unable to connect to AMQP server on %(broker)s: "
+                            "%(e)s. Sleeping %(delay)s seconds") % msg_dict
+                    LOG.error(msg)
+                    time.sleep(delay)
+                    delay = min(delay + 1, 5)
+            else:
+                LOG.info(_('Connected to AMQP server on %s'), broker['host'])
+                break
+
+        self.session = self.connection.session()
+
+        if self.consumers:
+            consumers = self.consumers
+            self.consumers = {}
+
+            for consumer in six.itervalues(consumers):
+                consumer.reconnect(self.session)
+                self._register_consumer(consumer)
+
+            LOG.debug("Re-established AMQP queues")
+
+    def ensure(self, error_callback, method, retry=None):
+        while True:
+            try:
+                return method()
+            except (qpid_exceptions.Empty,
+                    qpid_exceptions.MessagingError) as e:
+                if error_callback:
+                    error_callback(e)
+                self.reconnect(retry=retry)
+
+    def close(self):
+        """Close/release this connection."""
+        try:
+            self.connection.close()
+        except Exception:
+            # NOTE(dripton) Logging exceptions that happen during cleanup just
+            # causes confusion; there's really nothing useful we can do with
+            # them.
+            pass
+        self.connection = None
+
+    def reset(self):
+        """Reset a connection so it can be used again."""
+        self.session.close()
+        self.session = self.connection.session()
+        self.consumers = {}
+
+    def declare_consumer(self, consumer_cls, topic, callback):
+        """Create a Consumer using the class that was passed in and
+        add it to our list of consumers
+        """
+        def _connect_error(exc):
+            log_info = {'topic': topic, 'err_str': exc}
+            LOG.error(_("Failed to declare consumer for topic '%(topic)s': "
+                        "%(err_str)s"), log_info)
+
+        def _declare_consumer():
+            consumer = consumer_cls(self.conf, self.session, topic, callback)
+            self._register_consumer(consumer)
+            return consumer
+
+        return self.ensure(_connect_error, _declare_consumer)
+
+    def iterconsume(self, limit=None, timeout=None):
+        """Return an iterator that will consume from all queues/consumers."""
+
+        def _error_callback(exc):
+            if isinstance(exc, qpid_exceptions.Empty):
+                LOG.debug('Timed out waiting for RPC response: %s', exc)
+                raise rpc_common.Timeout()
+            else:
+                LOG.exception(_('Failed to consume message from queue: %s'),
+                              exc)
+
+        def _consume():
+            nxt_receiver = self.session.next_receiver(timeout=timeout)
+            try:
+                self._lookup_consumer(nxt_receiver).consume()
+            except Exception:
+                LOG.exception(_("Error processing message.  Skipping it."))
+
+        for iteration in itertools.count(0):
+            if limit and iteration >= limit:
+                raise StopIteration
+            yield self.ensure(_error_callback, _consume)
+
+    def publisher_send(self, cls, topic, msg, retry=None, **kwargs):
+        """Send to a publisher based on the publisher class."""
+
+        def _connect_error(exc):
+            log_info = {'topic': topic, 'err_str': exc}
+            LOG.exception(_("Failed to publish message to topic "
+                          "'%(topic)s': %(err_str)s"), log_info)
+
+        def _publisher_send():
+            publisher = cls(self.conf, self.session, topic=topic, **kwargs)
+            publisher.send(msg)
+
+        return self.ensure(_connect_error, _publisher_send, retry=retry)
+
+    def declare_direct_consumer(self, topic, callback):
+        """Create a 'direct' queue.
+        In nova's use, this is generally a msg_id queue used for
+        responses for call/multicall
+        """
+        self.declare_consumer(DirectConsumer, topic, callback)
+
+    def declare_topic_consumer(self, exchange_name, topic, callback=None,
+                               queue_name=None):
+        """Create a 'topic' consumer."""
+        self.declare_consumer(functools.partial(TopicConsumer,
+                                                name=queue_name,
+                                                exchange_name=exchange_name,
+                                                ),
+                              topic, callback)
+
+    def declare_fanout_consumer(self, topic, callback):
+        """Create a 'fanout' consumer."""
+        self.declare_consumer(FanoutConsumer, topic, callback)
+
+    def direct_send(self, msg_id, msg):
+        """Send a 'direct' message."""
+        self.publisher_send(DirectPublisher, topic=msg_id, msg=msg)
+
+    def topic_send(self, exchange_name, topic, msg, timeout=None, retry=None):
+        """Send a 'topic' message."""
+        #
+        # We want to create a message with attributes, for example a TTL. We
+        # don't really need to keep 'msg' in its JSON format any longer
+        # so let's create an actual Qpid message here and get some
+        # value-add on the go.
+        #
+        # WARNING: Request timeout happens to be in the same units as
+        # Qpid's TTL (seconds). If this changes in the future, then this
+        # will need to be altered accordingly.
+        #
+        qpid_message = qpid_messaging.Message(content=msg, ttl=timeout)
+        self.publisher_send(TopicPublisher, topic=topic, msg=qpid_message,
+                            exchange_name=exchange_name, retry=retry)
+
+    def fanout_send(self, topic, msg, retry=None):
+        """Send a 'fanout' message."""
+        self.publisher_send(FanoutPublisher, topic=topic, msg=msg, retry=retry)
+
+    def notify_send(self, exchange_name, topic, msg, retry=None, **kwargs):
+        """Send a notify message on a topic."""
+        self.publisher_send(NotifyPublisher, topic=topic, msg=msg,
+                            exchange_name=exchange_name, retry=retry)
+
+    def consume(self, limit=None, timeout=None):
+        """Consume from all queues/consumers."""
+        it = self.iterconsume(limit=limit, timeout=timeout)
+        while True:
+            try:
+                six.next(it)
+            except StopIteration:
+                return
+
+
+class QpidDriver(amqpdriver.AMQPDriverBase):
+
+    def __init__(self, conf, url,
+                 default_exchange=None, allowed_remote_exmods=None):
+        conf.register_opts(qpid_opts)
+        conf.register_opts(rpc_amqp.amqp_opts)
+
+        connection_pool = rpc_amqp.get_connection_pool(conf, url, Connection)
+
+        super(QpidDriver, self).__init__(conf, url,
+                                         connection_pool,
+                                         default_exchange,
+                                         allowed_remote_exmods)
 
=== added directory '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch'
=== added file '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/.timestamp'
=== added directory '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/oslo'
=== added directory '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/oslo/messaging'
=== added directory '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/oslo/messaging/_drivers'
=== added file '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/oslo/messaging/_drivers/impl_rabbit.py'
--- .pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/oslo/messaging/_drivers/impl_rabbit.py	1970-01-01 00:00:00 +0000
+++ .pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/oslo/messaging/_drivers/impl_rabbit.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,832 @@
+#    Copyright 2011 OpenStack Foundation
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+import itertools
+import logging
+import random
+import socket
+import ssl
+import time
+import uuid
+
+import kombu
+import kombu.connection
+import kombu.entity
+import kombu.messaging
+import six
+
+from oslo.config import cfg
+from oslo.messaging._drivers import amqp as rpc_amqp
+from oslo.messaging._drivers import amqpdriver
+from oslo.messaging._drivers import common as rpc_common
+from oslo.messaging import exceptions
+from oslo.messaging.openstack.common.gettextutils import _
+from oslo.utils import netutils
+
+rabbit_opts = [
+    cfg.StrOpt('kombu_ssl_version',
+               default='',
+               help='SSL version to use (valid only if SSL enabled). '
+                    'valid values are TLSv1, SSLv23 and SSLv3. SSLv2 may '
+                    'be available on some distributions.'
+               ),
+    cfg.StrOpt('kombu_ssl_keyfile',
+               default='',
+               help='SSL key file (valid only if SSL enabled).'),
+    cfg.StrOpt('kombu_ssl_certfile',
+               default='',
+               help='SSL cert file (valid only if SSL enabled).'),
+    cfg.StrOpt('kombu_ssl_ca_certs',
+               default='',
+               help='SSL certification authority file '
+                    '(valid only if SSL enabled).'),
+    cfg.FloatOpt('kombu_reconnect_delay',
+                 default=1.0,
+                 help='How long to wait before reconnecting in response to an '
+                      'AMQP consumer cancel notification.'),
+    cfg.StrOpt('rabbit_host',
+               default='localhost',
+               help='The RabbitMQ broker address where a single node is '
+                    'used.'),
+    cfg.IntOpt('rabbit_port',
+               default=5672,
+               help='The RabbitMQ broker port where a single node is used.'),
+    cfg.ListOpt('rabbit_hosts',
+                default=['$rabbit_host:$rabbit_port'],
+                help='RabbitMQ HA cluster host:port pairs.'),
+    cfg.BoolOpt('rabbit_use_ssl',
+                default=False,
+                help='Connect over SSL for RabbitMQ.'),
+    cfg.StrOpt('rabbit_userid',
+               default='guest',
+               help='The RabbitMQ userid.'),
+    cfg.StrOpt('rabbit_password',
+               default='guest',
+               help='The RabbitMQ password.',
+               secret=True),
+    cfg.StrOpt('rabbit_login_method',
+               default='AMQPLAIN',
+               help='the RabbitMQ login method'),
+    cfg.StrOpt('rabbit_virtual_host',
+               default='/',
+               help='The RabbitMQ virtual host.'),
+    cfg.IntOpt('rabbit_retry_interval',
+               default=1,
+               help='How frequently to retry connecting with RabbitMQ.'),
+    cfg.IntOpt('rabbit_retry_backoff',
+               default=2,
+               help='How long to backoff for between retries when connecting '
+                    'to RabbitMQ.'),
+    cfg.IntOpt('rabbit_max_retries',
+               default=0,
+               help='Maximum number of RabbitMQ connection retries. '
+                    'Default is 0 (infinite retry count).'),
+    cfg.BoolOpt('rabbit_ha_queues',
+                default=False,
+                help='Use HA queues in RabbitMQ (x-ha-policy: all). '
+                     'If you change this option, you must wipe the '
+                     'RabbitMQ database.'),
+
+    # FIXME(markmc): this was toplevel in openstack.common.rpc
+    cfg.BoolOpt('fake_rabbit',
+                default=False,
+                help='If passed, use a fake RabbitMQ provider.'),
+]
+
+LOG = logging.getLogger(__name__)
+
+
+def _get_queue_arguments(conf):
+    """Construct the arguments for declaring a queue.
+
+    If the rabbit_ha_queues option is set, we declare a mirrored queue
+    as described here:
+
+      http://www.rabbitmq.com/ha.html
+
+    Setting x-ha-policy to all means that the queue will be mirrored
+    to all nodes in the cluster.
+    """
+    return {'x-ha-policy': 'all'} if conf.rabbit_ha_queues else {}
+
+
+class RabbitMessage(dict):
+    def __init__(self, raw_message):
+        super(RabbitMessage, self).__init__(
+            rpc_common.deserialize_msg(raw_message.payload))
+        self._raw_message = raw_message
+
+    def acknowledge(self):
+        self._raw_message.ack()
+
+    def requeue(self):
+        self._raw_message.requeue()
+
+
+class ConsumerBase(object):
+    """Consumer base class."""
+
+    def __init__(self, channel, callback, tag, **kwargs):
+        """Declare a queue on an amqp channel.
+
+        'channel' is the amqp channel to use
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        queue name, exchange name, and other kombu options are
+        passed in here as a dictionary.
+        """
+        self.callback = callback
+        self.tag = six.text_type(tag)
+        self.kwargs = kwargs
+        self.queue = None
+        self.reconnect(channel)
+
+    def reconnect(self, channel):
+        """Re-declare the queue after a rabbit reconnect."""
+        self.channel = channel
+        self.kwargs['channel'] = channel
+        self.queue = kombu.entity.Queue(**self.kwargs)
+        self.queue.declare()
+
+    def _callback_handler(self, message, callback):
+        """Call callback with deserialized message.
+
+        Messages that are processed and ack'ed.
+        """
+
+        try:
+            callback(RabbitMessage(message))
+        except Exception:
+            LOG.exception(_("Failed to process message"
+                            " ... skipping it."))
+            message.ack()
+
+    def consume(self, *args, **kwargs):
+        """Actually declare the consumer on the amqp channel.  This will
+        start the flow of messages from the queue.  Using the
+        Connection.iterconsume() iterator will process the messages,
+        calling the appropriate callback.
+
+        If a callback is specified in kwargs, use that.  Otherwise,
+        use the callback passed during __init__()
+
+        If kwargs['nowait'] is True, then this call will block until
+        a message is read.
+
+        """
+
+        options = {'consumer_tag': self.tag}
+        options['nowait'] = kwargs.get('nowait', False)
+        callback = kwargs.get('callback', self.callback)
+        if not callback:
+            raise ValueError("No callback defined")
+
+        def _callback(raw_message):
+            message = self.channel.message_to_python(raw_message)
+            self._callback_handler(message, callback)
+
+        self.queue.consume(*args, callback=_callback, **options)
+
+    def cancel(self):
+        """Cancel the consuming from the queue, if it has started."""
+        try:
+            self.queue.cancel(self.tag)
+        except KeyError as e:
+            # NOTE(comstud): Kludge to get around a amqplib bug
+            if six.text_type(e) != "u'%s'" % self.tag:
+                raise
+        self.queue = None
+
+
+class DirectConsumer(ConsumerBase):
+    """Queue/consumer class for 'direct'."""
+
+    def __init__(self, conf, channel, msg_id, callback, tag, **kwargs):
+        """Init a 'direct' queue.
+
+        'channel' is the amqp channel to use
+        'msg_id' is the msg_id to listen on
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        Other kombu options may be passed
+        """
+        # Default options
+        options = {'durable': False,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=msg_id,
+                                         type='direct',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(DirectConsumer, self).__init__(channel,
+                                             callback,
+                                             tag,
+                                             name=msg_id,
+                                             exchange=exchange,
+                                             routing_key=msg_id,
+                                             **options)
+
+
+class TopicConsumer(ConsumerBase):
+    """Consumer class for 'topic'."""
+
+    def __init__(self, conf, channel, topic, callback, tag, exchange_name,
+                 name=None, **kwargs):
+        """Init a 'topic' queue.
+
+        :param channel: the amqp channel to use
+        :param topic: the topic to listen on
+        :paramtype topic: str
+        :param callback: the callback to call when messages are received
+        :param tag: a unique ID for the consumer on the channel
+        :param exchange_name: the exchange name to use
+        :param name: optional queue name, defaults to topic
+        :paramtype name: str
+
+        Other kombu options may be passed as keyword arguments
+        """
+        # Default options
+        options = {'durable': conf.amqp_durable_queues,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': conf.amqp_auto_delete,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=exchange_name,
+                                         type='topic',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(TopicConsumer, self).__init__(channel,
+                                            callback,
+                                            tag,
+                                            name=name or topic,
+                                            exchange=exchange,
+                                            routing_key=topic,
+                                            **options)
+
+
+class FanoutConsumer(ConsumerBase):
+    """Consumer class for 'fanout'."""
+
+    def __init__(self, conf, channel, topic, callback, tag, **kwargs):
+        """Init a 'fanout' queue.
+
+        'channel' is the amqp channel to use
+        'topic' is the topic to listen on
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        Other kombu options may be passed
+        """
+        unique = uuid.uuid4().hex
+        exchange_name = '%s_fanout' % topic
+        queue_name = '%s_fanout_%s' % (topic, unique)
+
+        # Default options
+        options = {'durable': False,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=exchange_name, type='fanout',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(FanoutConsumer, self).__init__(channel, callback, tag,
+                                             name=queue_name,
+                                             exchange=exchange,
+                                             routing_key=topic,
+                                             **options)
+
+
+class Publisher(object):
+    """Base Publisher class."""
+
+    def __init__(self, channel, exchange_name, routing_key, **kwargs):
+        """Init the Publisher class with the exchange_name, routing_key,
+        and other options
+        """
+        self.exchange_name = exchange_name
+        self.routing_key = routing_key
+        self.kwargs = kwargs
+        self.reconnect(channel)
+
+    def reconnect(self, channel):
+        """Re-establish the Producer after a rabbit reconnection."""
+        self.exchange = kombu.entity.Exchange(name=self.exchange_name,
+                                              **self.kwargs)
+        self.producer = kombu.messaging.Producer(exchange=self.exchange,
+                                                 channel=channel,
+                                                 routing_key=self.routing_key)
+
+    def send(self, msg, timeout=None):
+        """Send a message."""
+        if timeout:
+            #
+            # AMQP TTL is in milliseconds when set in the header.
+            #
+            self.producer.publish(msg, headers={'ttl': (timeout * 1000)})
+        else:
+            self.producer.publish(msg)
+
+
+class DirectPublisher(Publisher):
+    """Publisher class for 'direct'."""
+    def __init__(self, conf, channel, topic, **kwargs):
+        """Init a 'direct' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+
+        options = {'durable': False,
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(DirectPublisher, self).__init__(channel, topic, topic,
+                                              type='direct', **options)
+
+
+class TopicPublisher(Publisher):
+    """Publisher class for 'topic'."""
+    def __init__(self, conf, channel, exchange_name, topic, **kwargs):
+        """Init a 'topic' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+        options = {'durable': conf.amqp_durable_queues,
+                   'auto_delete': conf.amqp_auto_delete,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(TopicPublisher, self).__init__(channel,
+                                             exchange_name,
+                                             topic,
+                                             type='topic',
+                                             **options)
+
+
+class FanoutPublisher(Publisher):
+    """Publisher class for 'fanout'."""
+    def __init__(self, conf, channel, topic, **kwargs):
+        """Init a 'fanout' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+        options = {'durable': False,
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(FanoutPublisher, self).__init__(channel, '%s_fanout' % topic,
+                                              None, type='fanout', **options)
+
+
+class NotifyPublisher(TopicPublisher):
+    """Publisher class for 'notify'."""
+
+    def __init__(self, conf, channel, exchange_name, topic, **kwargs):
+        self.durable = kwargs.pop('durable', conf.amqp_durable_queues)
+        self.queue_arguments = _get_queue_arguments(conf)
+        super(NotifyPublisher, self).__init__(conf, channel, exchange_name,
+                                              topic, **kwargs)
+
+    def reconnect(self, channel):
+        super(NotifyPublisher, self).reconnect(channel)
+
+        # NOTE(jerdfelt): Normally the consumer would create the queue, but
+        # we do this to ensure that messages don't get dropped if the
+        # consumer is started after we do
+        queue = kombu.entity.Queue(channel=channel,
+                                   exchange=self.exchange,
+                                   durable=self.durable,
+                                   name=self.routing_key,
+                                   routing_key=self.routing_key,
+                                   queue_arguments=self.queue_arguments)
+        queue.declare()
+
+
+class Connection(object):
+    """Connection object."""
+
+    pools = {}
+
+    def __init__(self, conf, url):
+        self.consumers = []
+        self.conf = conf
+        self.max_retries = self.conf.rabbit_max_retries
+        # Try forever?
+        if self.max_retries <= 0:
+            self.max_retries = None
+        self.interval_start = self.conf.rabbit_retry_interval
+        self.interval_stepping = self.conf.rabbit_retry_backoff
+        # max retry-interval = 30 seconds
+        self.interval_max = 30
+        self.memory_transport = False
+
+        ssl_params = self._fetch_ssl_params()
+
+        if url.virtual_host is not None:
+            virtual_host = url.virtual_host
+        else:
+            virtual_host = self.conf.rabbit_virtual_host
+
+        self.brokers_params = []
+        if url.hosts:
+            for host in url.hosts:
+                params = {
+                    'hostname': host.hostname,
+                    'port': host.port or 5672,
+                    'userid': host.username or '',
+                    'password': host.password or '',
+                    'login_method': self.conf.rabbit_login_method,
+                    'virtual_host': virtual_host
+                }
+                if self.conf.fake_rabbit:
+                    params['transport'] = 'memory'
+                if self.conf.rabbit_use_ssl:
+                    params['ssl'] = ssl_params
+
+                self.brokers_params.append(params)
+        else:
+            # Old configuration format
+            for adr in self.conf.rabbit_hosts:
+                hostname, port = netutils.parse_host_port(
+                    adr, default_port=self.conf.rabbit_port)
+
+                params = {
+                    'hostname': hostname,
+                    'port': port,
+                    'userid': self.conf.rabbit_userid,
+                    'password': self.conf.rabbit_password,
+                    'login_method': self.conf.rabbit_login_method,
+                    'virtual_host': virtual_host
+                }
+
+                if self.conf.fake_rabbit:
+                    params['transport'] = 'memory'
+                if self.conf.rabbit_use_ssl:
+                    params['ssl'] = ssl_params
+
+                self.brokers_params.append(params)
+
+        random.shuffle(self.brokers_params)
+        self.brokers = itertools.cycle(self.brokers_params)
+
+        self.memory_transport = self.conf.fake_rabbit
+
+        self.connection = None
+        self.do_consume = None
+        self.reconnect()
+
+    # FIXME(markmc): use oslo sslutils when it is available as a library
+    _SSL_PROTOCOLS = {
+        "tlsv1": ssl.PROTOCOL_TLSv1,
+        "sslv23": ssl.PROTOCOL_SSLv23,
+        "sslv3": ssl.PROTOCOL_SSLv3
+    }
+
+    try:
+        _SSL_PROTOCOLS["sslv2"] = ssl.PROTOCOL_SSLv2
+    except AttributeError:
+        pass
+
+    @classmethod
+    def validate_ssl_version(cls, version):
+        key = version.lower()
+        try:
+            return cls._SSL_PROTOCOLS[key]
+        except KeyError:
+            raise RuntimeError(_("Invalid SSL version : %s") % version)
+
+    def _fetch_ssl_params(self):
+        """Handles fetching what ssl params should be used for the connection
+        (if any).
+        """
+        ssl_params = dict()
+
+        # http://docs.python.org/library/ssl.html - ssl.wrap_socket
+        if self.conf.kombu_ssl_version:
+            ssl_params['ssl_version'] = self.validate_ssl_version(
+                self.conf.kombu_ssl_version)
+        if self.conf.kombu_ssl_keyfile:
+            ssl_params['keyfile'] = self.conf.kombu_ssl_keyfile
+        if self.conf.kombu_ssl_certfile:
+            ssl_params['certfile'] = self.conf.kombu_ssl_certfile
+        if self.conf.kombu_ssl_ca_certs:
+            ssl_params['ca_certs'] = self.conf.kombu_ssl_ca_certs
+            # We might want to allow variations in the
+            # future with this?
+            ssl_params['cert_reqs'] = ssl.CERT_REQUIRED
+
+        # Return the extended behavior or just have the default behavior
+        return ssl_params or True
+
+    def _connect(self, broker):
+        """Connect to rabbit.  Re-establish any queues that may have
+        been declared before if we are reconnecting.  Exceptions should
+        be handled by the caller.
+        """
+        LOG.info(_("Connecting to AMQP server on "
+                   "%(hostname)s:%(port)d"), broker)
+        self.connection = kombu.connection.BrokerConnection(**broker)
+        self.connection_errors = self.connection.connection_errors
+        self.channel_errors = self.connection.channel_errors
+        if self.memory_transport:
+            # Kludge to speed up tests.
+            self.connection.transport.polling_interval = 0.0
+        self.do_consume = True
+        self.consumer_num = itertools.count(1)
+        self.connection.connect()
+        self.channel = self.connection.channel()
+        # work around 'memory' transport bug in 1.1.3
+        if self.memory_transport:
+            self.channel._new_queue('ae.undeliver')
+        for consumer in self.consumers:
+            consumer.reconnect(self.channel)
+        LOG.info(_('Connected to AMQP server on %(hostname)s:%(port)d'),
+                 broker)
+
+    def _disconnect(self):
+        if self.connection:
+            # XXX(nic): when reconnecting to a RabbitMQ cluster
+            # with mirrored queues in use, the attempt to release the
+            # connection can hang "indefinitely" somewhere deep down
+            # in Kombu.  Blocking the thread for a bit prior to
+            # release seems to kludge around the problem where it is
+            # otherwise reproduceable.
+            if self.conf.kombu_reconnect_delay > 0:
+                LOG.info(_("Delaying reconnect for %1.1f seconds...") %
+                         self.conf.kombu_reconnect_delay)
+                time.sleep(self.conf.kombu_reconnect_delay)
+
+            try:
+                self.connection.release()
+            except self.connection_errors:
+                pass
+            self.connection = None
+
+    def reconnect(self, retry=None):
+        """Handles reconnecting and re-establishing queues.
+        Will retry up to retry number of times.
+        retry = None means use the value of rabbit_max_retries
+        retry = -1 means to retry forever
+        retry = 0 means no retry
+        retry = N means N retries
+        Sleep between tries, starting at self.interval_start
+        seconds, backing off self.interval_stepping number of seconds
+        each attempt.
+        """
+
+        attempt = 0
+        loop_forever = False
+        if retry is None:
+            retry = self.max_retries
+        if retry is None or retry < 0:
+            loop_forever = True
+
+        while True:
+            self._disconnect()
+
+            broker = six.next(self.brokers)
+            attempt += 1
+            try:
+                self._connect(broker)
+                return
+            except IOError as ex:
+                e = ex
+            except self.connection_errors as ex:
+                e = ex
+            except Exception as ex:
+                # NOTE(comstud): Unfortunately it's possible for amqplib
+                # to return an error not covered by its transport
+                # connection_errors in the case of a timeout waiting for
+                # a protocol response.  (See paste link in LP888621)
+                # So, we check all exceptions for 'timeout' in them
+                # and try to reconnect in this case.
+                if 'timeout' not in six.text_type(e):
+                    raise
+                e = ex
+
+            log_info = {}
+            log_info['err_str'] = e
+            log_info['retry'] = retry or 0
+            log_info.update(broker)
+
+            if not loop_forever and attempt > retry:
+                msg = _('Unable to connect to AMQP server on '
+                        '%(hostname)s:%(port)d after %(retry)d '
+                        'tries: %(err_str)s') % log_info
+                LOG.error(msg)
+                raise exceptions.MessageDeliveryFailure(msg)
+            else:
+                if attempt == 1:
+                    sleep_time = self.interval_start or 1
+                elif attempt > 1:
+                    sleep_time += self.interval_stepping
+
+                sleep_time = min(sleep_time, self.interval_max)
+
+                log_info['sleep_time'] = sleep_time
+                if 'Socket closed' in six.text_type(e):
+                    LOG.error(_('AMQP server %(hostname)s:%(port)d closed'
+                                ' the connection. Check login credentials:'
+                                ' %(err_str)s'), log_info)
+                else:
+                    LOG.error(_('AMQP server on %(hostname)s:%(port)d is '
+                                'unreachable: %(err_str)s. Trying again in '
+                                '%(sleep_time)d seconds.'), log_info)
+                time.sleep(sleep_time)
+
+    def ensure(self, error_callback, method, retry=None):
+        while True:
+            try:
+                return method()
+            except self.connection_errors as e:
+                if error_callback:
+                    error_callback(e)
+            except self.channel_errors as e:
+                if error_callback:
+                    error_callback(e)
+            except (socket.timeout, IOError) as e:
+                if error_callback:
+                    error_callback(e)
+            except Exception as e:
+                # NOTE(comstud): Unfortunately it's possible for amqplib
+                # to return an error not covered by its transport
+                # connection_errors in the case of a timeout waiting for
+                # a protocol response.  (See paste link in LP888621)
+                # So, we check all exceptions for 'timeout' in them
+                # and try to reconnect in this case.
+                if 'timeout' not in six.text_type(e):
+                    raise
+                if error_callback:
+                    error_callback(e)
+            self.reconnect(retry=retry)
+
+    def get_channel(self):
+        """Convenience call for bin/clear_rabbit_queues."""
+        return self.channel
+
+    def close(self):
+        """Close/release this connection."""
+        if self.connection:
+            self.connection.release()
+            self.connection = None
+
+    def reset(self):
+        """Reset a connection so it can be used again."""
+        self.channel.close()
+        self.channel = self.connection.channel()
+        # work around 'memory' transport bug in 1.1.3
+        if self.memory_transport:
+            self.channel._new_queue('ae.undeliver')
+        self.consumers = []
+
+    def declare_consumer(self, consumer_cls, topic, callback):
+        """Create a Consumer using the class that was passed in and
+        add it to our list of consumers
+        """
+
+        def _connect_error(exc):
+            log_info = {'topic': topic, 'err_str': exc}
+            LOG.error(_("Failed to declare consumer for topic '%(topic)s': "
+                      "%(err_str)s"), log_info)
+
+        def _declare_consumer():
+            consumer = consumer_cls(self.conf, self.channel, topic, callback,
+                                    six.next(self.consumer_num))
+            self.consumers.append(consumer)
+            return consumer
+
+        return self.ensure(_connect_error, _declare_consumer)
+
+    def iterconsume(self, limit=None, timeout=None):
+        """Return an iterator that will consume from all queues/consumers."""
+
+        timer = rpc_common.DecayingTimer(duration=timeout)
+        timer.start()
+
+        def _raise_timeout(exc):
+            LOG.debug('Timed out waiting for RPC response: %s', exc)
+            raise rpc_common.Timeout()
+
+        def _error_callback(exc):
+            timer.check_return(_raise_timeout, exc)
+            LOG.exception(_('Failed to consume message from queue: %s'),
+                          exc)
+            self.do_consume = True
+
+        def _consume():
+            if self.do_consume:
+                queues_head = self.consumers[:-1]  # not fanout.
+                queues_tail = self.consumers[-1]  # fanout
+                for queue in queues_head:
+                    queue.consume(nowait=True)
+                queues_tail.consume(nowait=False)
+                self.do_consume = False
+
+            poll_timeout = 1 if timeout is None else min(timeout, 1)
+            while True:
+                try:
+                    return self.connection.drain_events(timeout=poll_timeout)
+                except socket.timeout as exc:
+                    poll_timeout = timer.check_return(_raise_timeout, exc,
+                                                      maximum=1)
+
+        for iteration in itertools.count(0):
+            if limit and iteration >= limit:
+                raise StopIteration
+            yield self.ensure(_error_callback, _consume)
+
+    def publisher_send(self, cls, topic, msg, timeout=None, retry=None,
+                       **kwargs):
+        """Send to a publisher based on the publisher class."""
+
+        def _error_callback(exc):
+            log_info = {'topic': topic, 'err_str': exc}
+            LOG.exception(_("Failed to publish message to topic "
+                          "'%(topic)s': %(err_str)s"), log_info)
+
+        def _publish():
+            publisher = cls(self.conf, self.channel, topic=topic, **kwargs)
+            publisher.send(msg, timeout)
+
+        self.ensure(_error_callback, _publish, retry=retry)
+
+    def declare_direct_consumer(self, topic, callback):
+        """Create a 'direct' queue.
+        In nova's use, this is generally a msg_id queue used for
+        responses for call/multicall
+        """
+        self.declare_consumer(DirectConsumer, topic, callback)
+
+    def declare_topic_consumer(self, exchange_name, topic, callback=None,
+                               queue_name=None):
+        """Create a 'topic' consumer."""
+        self.declare_consumer(functools.partial(TopicConsumer,
+                                                name=queue_name,
+                                                exchange_name=exchange_name,
+                                                ),
+                              topic, callback)
+
+    def declare_fanout_consumer(self, topic, callback):
+        """Create a 'fanout' consumer."""
+        self.declare_consumer(FanoutConsumer, topic, callback)
+
+    def direct_send(self, msg_id, msg):
+        """Send a 'direct' message."""
+        self.publisher_send(DirectPublisher, msg_id, msg)
+
+    def topic_send(self, exchange_name, topic, msg, timeout=None, retry=None):
+        """Send a 'topic' message."""
+        self.publisher_send(TopicPublisher, topic, msg, timeout,
+                            exchange_name=exchange_name, retry=retry)
+
+    def fanout_send(self, topic, msg, retry=None):
+        """Send a 'fanout' message."""
+        self.publisher_send(FanoutPublisher, topic, msg, retry=retry)
+
+    def notify_send(self, exchange_name, topic, msg, retry=None, **kwargs):
+        """Send a notify message on a topic."""
+        self.publisher_send(NotifyPublisher, topic, msg, timeout=None,
+                            exchange_name=exchange_name, retry=retry, **kwargs)
+
+    def consume(self, limit=None, timeout=None):
+        """Consume from all queues/consumers."""
+        it = self.iterconsume(limit=limit, timeout=timeout)
+        while True:
+            try:
+                six.next(it)
+            except StopIteration:
+                return
+
+
+class RabbitDriver(amqpdriver.AMQPDriverBase):
+
+    def __init__(self, conf, url,
+                 default_exchange=None,
+                 allowed_remote_exmods=None):
+        conf.register_opts(rabbit_opts)
+        conf.register_opts(rpc_amqp.amqp_opts)
+
+        connection_pool = rpc_amqp.get_connection_pool(conf, url, Connection)
+
+        super(RabbitDriver, self).__init__(conf, url,
+                                           connection_pool,
+                                           default_exchange,
+                                           allowed_remote_exmods)
+
+    def require_features(self, requeue=True):
+        pass
 
=== added directory '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/tests'
=== added directory '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/tests/drivers'
=== added file '.pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/tests/drivers/test_impl_rabbit.py'
--- .pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/tests/drivers/test_impl_rabbit.py	1970-01-01 00:00:00 +0000
+++ .pc/0004-rabbit-redeclare-consumers-when-ack-requeue-fail.patch/tests/drivers/test_impl_rabbit.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,735 @@
+# Copyright 2013 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import datetime
+import operator
+import sys
+import threading
+import uuid
+
+import fixtures
+import kombu
+import mock
+import testscenarios
+
+from oslo import messaging
+from oslo.messaging._drivers import amqpdriver
+from oslo.messaging._drivers import common as driver_common
+from oslo.messaging._drivers import impl_rabbit as rabbit_driver
+from oslo.messaging.openstack.common import jsonutils
+from tests import utils as test_utils
+
+load_tests = testscenarios.load_tests_apply_scenarios
+
+
+class TestRabbitDriverLoad(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(TestRabbitDriverLoad, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+    def test_driver_load(self):
+        transport = messaging.get_transport(self.conf)
+        self.assertIsInstance(transport._driver, rabbit_driver.RabbitDriver)
+
+
+class TestRabbitTransportURL(test_utils.BaseTestCase):
+
+    scenarios = [
+        ('none', dict(url=None,
+                      expected=[dict(hostname='localhost',
+                                     port=5672,
+                                     userid='guest',
+                                     password='guest',
+                                     virtual_host='/')])),
+        ('empty',
+         dict(url='rabbit:///',
+              expected=[dict(hostname='localhost',
+                             port=5672,
+                             userid='guest',
+                             password='guest',
+                             virtual_host='')])),
+        ('localhost',
+         dict(url='rabbit://localhost/',
+              expected=[dict(hostname='localhost',
+                             port=5672,
+                             userid='',
+                             password='',
+                             virtual_host='')])),
+        ('virtual_host',
+         dict(url='rabbit:///vhost',
+              expected=[dict(hostname='localhost',
+                             port=5672,
+                             userid='guest',
+                             password='guest',
+                             virtual_host='vhost')])),
+        ('no_creds',
+         dict(url='rabbit://host/virtual_host',
+              expected=[dict(hostname='host',
+                             port=5672,
+                             userid='',
+                             password='',
+                             virtual_host='virtual_host')])),
+        ('no_port',
+         dict(url='rabbit://user:password@host/virtual_host',
+              expected=[dict(hostname='host',
+                             port=5672,
+                             userid='user',
+                             password='password',
+                             virtual_host='virtual_host')])),
+        ('full_url',
+         dict(url='rabbit://user:password@host:10/virtual_host',
+              expected=[dict(hostname='host',
+                             port=10,
+                             userid='user',
+                             password='password',
+                             virtual_host='virtual_host')])),
+        ('full_two_url',
+         dict(url='rabbit://user:password@host:10,'
+              'user2:password2@host2:12/virtual_host',
+              expected=[dict(hostname='host',
+                             port=10,
+                             userid='user',
+                             password='password',
+                             virtual_host='virtual_host'),
+                        dict(hostname='host2',
+                             port=12,
+                             userid='user2',
+                             password='password2',
+                             virtual_host='virtual_host')
+                        ]
+              )),
+
+    ]
+
+    def test_transport_url(self):
+        self.messaging_conf.in_memory = True
+
+        transport = messaging.get_transport(self.conf, self.url)
+        self.addCleanup(transport.cleanup)
+        driver = transport._driver
+
+        brokers_params = driver._get_connection().brokers_params[:]
+        brokers_params = [dict((k, v) for k, v in broker.items()
+                               if k not in ['transport', 'login_method'])
+                          for broker in brokers_params]
+
+        self.assertEqual(sorted(self.expected,
+                                key=operator.itemgetter('hostname')),
+                         sorted(brokers_params,
+                                key=operator.itemgetter('hostname')))
+
+
+class TestSendReceive(test_utils.BaseTestCase):
+
+    _n_senders = [
+        ('single_sender', dict(n_senders=1)),
+        ('multiple_senders', dict(n_senders=10)),
+    ]
+
+    _context = [
+        ('empty_context', dict(ctxt={})),
+        ('with_context', dict(ctxt={'user': 'mark'})),
+    ]
+
+    _reply = [
+        ('rx_id', dict(rx_id=True, reply=None)),
+        ('none', dict(rx_id=False, reply=None)),
+        ('empty_list', dict(rx_id=False, reply=[])),
+        ('empty_dict', dict(rx_id=False, reply={})),
+        ('false', dict(rx_id=False, reply=False)),
+        ('zero', dict(rx_id=False, reply=0)),
+    ]
+
+    _failure = [
+        ('success', dict(failure=False)),
+        ('failure', dict(failure=True, expected=False)),
+        ('expected_failure', dict(failure=True, expected=True)),
+    ]
+
+    _timeout = [
+        ('no_timeout', dict(timeout=None)),
+        ('timeout', dict(timeout=0.01)),  # FIXME(markmc): timeout=0 is broken?
+    ]
+
+    @classmethod
+    def generate_scenarios(cls):
+        cls.scenarios = testscenarios.multiply_scenarios(cls._n_senders,
+                                                         cls._context,
+                                                         cls._reply,
+                                                         cls._failure,
+                                                         cls._timeout)
+
+    def setUp(self):
+        super(TestSendReceive, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+    def test_send_receive(self):
+        transport = messaging.get_transport(self.conf)
+        self.addCleanup(transport.cleanup)
+
+        driver = transport._driver
+
+        target = messaging.Target(topic='testtopic')
+
+        listener = driver.listen(target)
+
+        senders = []
+        replies = []
+        msgs = []
+        errors = []
+
+        def stub_error(msg, *a, **kw):
+            if (a and len(a) == 1 and isinstance(a[0], dict) and a[0]):
+                a = a[0]
+            errors.append(str(msg) % a)
+
+        self.stubs.Set(driver_common.LOG, 'error', stub_error)
+
+        def send_and_wait_for_reply(i):
+            try:
+                replies.append(driver.send(target,
+                                           self.ctxt,
+                                           {'tx_id': i},
+                                           wait_for_reply=True,
+                                           timeout=self.timeout))
+                self.assertFalse(self.failure)
+                self.assertIsNone(self.timeout)
+            except (ZeroDivisionError, messaging.MessagingTimeout) as e:
+                replies.append(e)
+                self.assertTrue(self.failure or self.timeout is not None)
+
+        while len(senders) < self.n_senders:
+            senders.append(threading.Thread(target=send_and_wait_for_reply,
+                                            args=(len(senders), )))
+
+        for i in range(len(senders)):
+            senders[i].start()
+
+            received = listener.poll()
+            self.assertIsNotNone(received)
+            self.assertEqual(self.ctxt, received.ctxt)
+            self.assertEqual({'tx_id': i}, received.message)
+            msgs.append(received)
+
+        # reply in reverse, except reply to the first guy second from last
+        order = list(range(len(senders) - 1, -1, -1))
+        if len(order) > 1:
+            order[-1], order[-2] = order[-2], order[-1]
+
+        for i in order:
+            if self.timeout is None:
+                if self.failure:
+                    try:
+                        raise ZeroDivisionError
+                    except Exception:
+                        failure = sys.exc_info()
+                    msgs[i].reply(failure=failure,
+                                  log_failure=not self.expected)
+                elif self.rx_id:
+                    msgs[i].reply({'rx_id': i})
+                else:
+                    msgs[i].reply(self.reply)
+            senders[i].join()
+
+        self.assertEqual(len(senders), len(replies))
+        for i, reply in enumerate(replies):
+            if self.timeout is not None:
+                self.assertIsInstance(reply, messaging.MessagingTimeout)
+            elif self.failure:
+                self.assertIsInstance(reply, ZeroDivisionError)
+            elif self.rx_id:
+                self.assertEqual({'rx_id': order[i]}, reply)
+            else:
+                self.assertEqual(self.reply, reply)
+
+        if not self.timeout and self.failure and not self.expected:
+            self.assertTrue(len(errors) > 0, errors)
+        else:
+            self.assertEqual(0, len(errors), errors)
+
+
+TestSendReceive.generate_scenarios()
+
+
+class TestPollAsync(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(TestPollAsync, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+    def test_poll_timeout(self):
+        transport = messaging.get_transport(self.conf)
+        self.addCleanup(transport.cleanup)
+        driver = transport._driver
+        target = messaging.Target(topic='testtopic')
+        listener = driver.listen(target)
+        received = listener.poll(timeout=0.050)
+        self.assertIsNone(received)
+
+
+class TestRacyWaitForReply(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(TestRacyWaitForReply, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+    def test_send_receive(self):
+        transport = messaging.get_transport(self.conf)
+        self.addCleanup(transport.cleanup)
+
+        driver = transport._driver
+
+        target = messaging.Target(topic='testtopic')
+
+        listener = driver.listen(target)
+
+        senders = []
+        replies = []
+        msgs = []
+
+        wait_conditions = []
+        orig_reply_waiter = amqpdriver.ReplyWaiter.wait
+
+        def reply_waiter(self, msg_id, timeout):
+            if wait_conditions:
+                cond = wait_conditions.pop()
+                with cond:
+                    cond.notify()
+                with cond:
+                    cond.wait()
+            return orig_reply_waiter(self, msg_id, timeout)
+
+        self.stubs.Set(amqpdriver.ReplyWaiter, 'wait', reply_waiter)
+
+        def send_and_wait_for_reply(i, wait_for_reply):
+            replies.append(driver.send(target,
+                                       {},
+                                       {'tx_id': i},
+                                       wait_for_reply=wait_for_reply,
+                                       timeout=None))
+
+        while len(senders) < 2:
+            t = threading.Thread(target=send_and_wait_for_reply,
+                                 args=(len(senders), True))
+            t.daemon = True
+            senders.append(t)
+
+        # test the case then msg_id is not set
+        t = threading.Thread(target=send_and_wait_for_reply,
+                             args=(len(senders), False))
+        t.daemon = True
+        senders.append(t)
+
+        # Start the first guy, receive his message, but delay his polling
+        notify_condition = threading.Condition()
+        wait_conditions.append(notify_condition)
+        with notify_condition:
+            senders[0].start()
+            notify_condition.wait()
+
+        msgs.append(listener.poll())
+        self.assertEqual({'tx_id': 0}, msgs[-1].message)
+
+        # Start the second guy, receive his message
+        senders[1].start()
+
+        msgs.append(listener.poll())
+        self.assertEqual({'tx_id': 1}, msgs[-1].message)
+
+        # Reply to both in order, making the second thread queue
+        # the reply meant for the first thread
+        msgs[0].reply({'rx_id': 0})
+        msgs[1].reply({'rx_id': 1})
+
+        # Wait for the second thread to finish
+        senders[1].join()
+
+        # Start the 3rd guy, receive his message
+        senders[2].start()
+
+        msgs.append(listener.poll())
+        self.assertEqual({'tx_id': 2}, msgs[-1].message)
+
+        # Verify the _send_reply was not invoked by driver:
+        with mock.patch.object(msgs[2], '_send_reply') as method:
+            msgs[2].reply({'rx_id': 2})
+            self.assertEqual(method.call_count, 0)
+
+        # Wait for the 3rd thread to finish
+        senders[2].join()
+
+        # Let the first thread continue
+        with notify_condition:
+            notify_condition.notify()
+
+        # Wait for the first thread to finish
+        senders[0].join()
+
+        # Verify replies were received out of order
+        self.assertEqual(len(senders), len(replies))
+        self.assertEqual({'rx_id': 1}, replies[0])
+        self.assertIsNone(replies[1])
+        self.assertEqual({'rx_id': 0}, replies[2])
+
+
+def _declare_queue(target):
+    connection = kombu.connection.BrokerConnection(transport='memory')
+
+    # Kludge to speed up tests.
+    connection.transport.polling_interval = 0.0
+
+    connection.connect()
+    channel = connection.channel()
+
+    # work around 'memory' transport bug in 1.1.3
+    channel._new_queue('ae.undeliver')
+
+    if target.fanout:
+        exchange = kombu.entity.Exchange(name=target.topic + '_fanout',
+                                         type='fanout',
+                                         durable=False,
+                                         auto_delete=True)
+        queue = kombu.entity.Queue(name=target.topic + '_fanout_12345',
+                                   channel=channel,
+                                   exchange=exchange,
+                                   routing_key=target.topic)
+    if target.server:
+        exchange = kombu.entity.Exchange(name='openstack',
+                                         type='topic',
+                                         durable=False,
+                                         auto_delete=False)
+        topic = '%s.%s' % (target.topic, target.server)
+        queue = kombu.entity.Queue(name=topic,
+                                   channel=channel,
+                                   exchange=exchange,
+                                   routing_key=topic)
+    else:
+        exchange = kombu.entity.Exchange(name='openstack',
+                                         type='topic',
+                                         durable=False,
+                                         auto_delete=False)
+        queue = kombu.entity.Queue(name=target.topic,
+                                   channel=channel,
+                                   exchange=exchange,
+                                   routing_key=target.topic)
+
+    queue.declare()
+
+    return connection, channel, queue
+
+
+class TestRequestWireFormat(test_utils.BaseTestCase):
+
+    _target = [
+        ('topic_target',
+         dict(topic='testtopic', server=None, fanout=False)),
+        ('server_target',
+         dict(topic='testtopic', server='testserver', fanout=False)),
+        # NOTE(markmc): https://github.com/celery/kombu/issues/195
+        ('fanout_target',
+         dict(topic='testtopic', server=None, fanout=True,
+              skip_msg='Requires kombu>2.5.12 to fix kombu issue #195')),
+    ]
+
+    _msg = [
+        ('empty_msg',
+         dict(msg={}, expected={})),
+        ('primitive_msg',
+         dict(msg={'foo': 'bar'}, expected={'foo': 'bar'})),
+        ('complex_msg',
+         dict(msg={'a': {'b': datetime.datetime(1920, 2, 3, 4, 5, 6, 7)}},
+              expected={'a': {'b': '1920-02-03T04:05:06.000007'}})),
+    ]
+
+    _context = [
+        ('empty_ctxt', dict(ctxt={}, expected_ctxt={})),
+        ('user_project_ctxt',
+         dict(ctxt={'user': 'mark', 'project': 'snarkybunch'},
+              expected_ctxt={'_context_user': 'mark',
+                             '_context_project': 'snarkybunch'})),
+    ]
+
+    @classmethod
+    def generate_scenarios(cls):
+        cls.scenarios = testscenarios.multiply_scenarios(cls._msg,
+                                                         cls._context,
+                                                         cls._target)
+
+    def setUp(self):
+        super(TestRequestWireFormat, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+        self.uuids = []
+        self.orig_uuid4 = uuid.uuid4
+        self.useFixture(fixtures.MonkeyPatch('uuid.uuid4', self.mock_uuid4))
+
+    def mock_uuid4(self):
+        self.uuids.append(self.orig_uuid4())
+        return self.uuids[-1]
+
+    def test_request_wire_format(self):
+        if hasattr(self, 'skip_msg'):
+            self.skipTest(self.skip_msg)
+
+        transport = messaging.get_transport(self.conf)
+        self.addCleanup(transport.cleanup)
+
+        driver = transport._driver
+
+        target = messaging.Target(topic=self.topic,
+                                  server=self.server,
+                                  fanout=self.fanout)
+
+        connection, channel, queue = _declare_queue(target)
+        self.addCleanup(connection.release)
+
+        driver.send(target, self.ctxt, self.msg)
+
+        msgs = []
+
+        def callback(msg):
+            msg = channel.message_to_python(msg)
+            msg.ack()
+            msgs.append(msg.payload)
+
+        queue.consume(callback=callback,
+                      consumer_tag='1',
+                      nowait=False)
+
+        connection.drain_events()
+
+        self.assertEqual(1, len(msgs))
+        self.assertIn('oslo.message', msgs[0])
+
+        received = msgs[0]
+        received['oslo.message'] = jsonutils.loads(received['oslo.message'])
+
+        # FIXME(markmc): add _msg_id and _reply_q check
+        expected_msg = {
+            '_unique_id': self.uuids[0].hex,
+        }
+        expected_msg.update(self.expected)
+        expected_msg.update(self.expected_ctxt)
+
+        expected = {
+            'oslo.version': '2.0',
+            'oslo.message': expected_msg,
+        }
+
+        self.assertEqual(expected, received)
+
+
+TestRequestWireFormat.generate_scenarios()
+
+
+def _create_producer(target):
+    connection = kombu.connection.BrokerConnection(transport='memory')
+
+    # Kludge to speed up tests.
+    connection.transport.polling_interval = 0.0
+
+    connection.connect()
+    channel = connection.channel()
+
+    # work around 'memory' transport bug in 1.1.3
+    channel._new_queue('ae.undeliver')
+
+    if target.fanout:
+        exchange = kombu.entity.Exchange(name=target.topic + '_fanout',
+                                         type='fanout',
+                                         durable=False,
+                                         auto_delete=True)
+        producer = kombu.messaging.Producer(exchange=exchange,
+                                            channel=channel,
+                                            routing_key=target.topic)
+    elif target.server:
+        exchange = kombu.entity.Exchange(name='openstack',
+                                         type='topic',
+                                         durable=False,
+                                         auto_delete=False)
+        topic = '%s.%s' % (target.topic, target.server)
+        producer = kombu.messaging.Producer(exchange=exchange,
+                                            channel=channel,
+                                            routing_key=topic)
+    else:
+        exchange = kombu.entity.Exchange(name='openstack',
+                                         type='topic',
+                                         durable=False,
+                                         auto_delete=False)
+        producer = kombu.messaging.Producer(exchange=exchange,
+                                            channel=channel,
+                                            routing_key=target.topic)
+
+    return connection, producer
+
+
+class TestReplyWireFormat(test_utils.BaseTestCase):
+
+    _target = [
+        ('topic_target',
+         dict(topic='testtopic', server=None, fanout=False)),
+        ('server_target',
+         dict(topic='testtopic', server='testserver', fanout=False)),
+        # NOTE(markmc): https://github.com/celery/kombu/issues/195
+        ('fanout_target',
+         dict(topic='testtopic', server=None, fanout=True,
+              skip_msg='Requires kombu>2.5.12 to fix kombu issue #195')),
+    ]
+
+    _msg = [
+        ('empty_msg',
+         dict(msg={}, expected={})),
+        ('primitive_msg',
+         dict(msg={'foo': 'bar'}, expected={'foo': 'bar'})),
+        ('complex_msg',
+         dict(msg={'a': {'b': '1920-02-03T04:05:06.000007'}},
+              expected={'a': {'b': '1920-02-03T04:05:06.000007'}})),
+    ]
+
+    _context = [
+        ('empty_ctxt', dict(ctxt={}, expected_ctxt={})),
+        ('user_project_ctxt',
+         dict(ctxt={'_context_user': 'mark',
+                    '_context_project': 'snarkybunch'},
+              expected_ctxt={'user': 'mark', 'project': 'snarkybunch'})),
+    ]
+
+    @classmethod
+    def generate_scenarios(cls):
+        cls.scenarios = testscenarios.multiply_scenarios(cls._msg,
+                                                         cls._context,
+                                                         cls._target)
+
+    def setUp(self):
+        super(TestReplyWireFormat, self).setUp()
+        self.messaging_conf.transport_driver = 'rabbit'
+        self.messaging_conf.in_memory = True
+
+    def test_reply_wire_format(self):
+        if hasattr(self, 'skip_msg'):
+            self.skipTest(self.skip_msg)
+
+        transport = messaging.get_transport(self.conf)
+        self.addCleanup(transport.cleanup)
+
+        driver = transport._driver
+
+        target = messaging.Target(topic=self.topic,
+                                  server=self.server,
+                                  fanout=self.fanout)
+
+        listener = driver.listen(target)
+
+        connection, producer = _create_producer(target)
+        self.addCleanup(connection.release)
+
+        msg = {
+            'oslo.version': '2.0',
+            'oslo.message': {}
+        }
+
+        msg['oslo.message'].update(self.msg)
+        msg['oslo.message'].update(self.ctxt)
+
+        msg['oslo.message'].update({
+            '_msg_id': uuid.uuid4().hex,
+            '_unique_id': uuid.uuid4().hex,
+            '_reply_q': 'reply_' + uuid.uuid4().hex,
+        })
+
+        msg['oslo.message'] = jsonutils.dumps(msg['oslo.message'])
+
+        producer.publish(msg)
+
+        received = listener.poll()
+        self.assertIsNotNone(received)
+        self.assertEqual(self.expected_ctxt, received.ctxt)
+        self.assertEqual(self.expected, received.message)
+
+
+TestReplyWireFormat.generate_scenarios()
+
+
+class RpcKombuHATestCase(test_utils.BaseTestCase):
+
+    def setUp(self):
+        super(RpcKombuHATestCase, self).setUp()
+        self.brokers = ['host1', 'host2', 'host3', 'host4', 'host5']
+        self.config(rabbit_hosts=self.brokers)
+
+        hostname_sets = set()
+        self.info = {'attempt': 0,
+                     'fail': False}
+
+        def _connect(myself, params):
+            # do as little work that is enough to pass connection attempt
+            myself.connection = kombu.connection.BrokerConnection(**params)
+            myself.connection_errors = myself.connection.connection_errors
+
+            hostname = params['hostname']
+            self.assertNotIn(hostname, hostname_sets)
+            hostname_sets.add(hostname)
+
+            self.info['attempt'] += 1
+            if self.info['fail']:
+                raise IOError('fake fail')
+
+        # just make sure connection instantiation does not fail with an
+        # exception
+        self.stubs.Set(rabbit_driver.Connection, '_connect', _connect)
+
+        # starting from the first broker in the list
+        url = messaging.TransportURL.parse(self.conf, None)
+        self.connection = rabbit_driver.Connection(self.conf, url)
+        self.addCleanup(self.connection.close)
+
+        self.info.update({'attempt': 0,
+                          'fail': True})
+        hostname_sets.clear()
+
+    def test_reconnect_order(self):
+        self.assertRaises(messaging.MessageDeliveryFailure,
+                          self.connection.reconnect,
+                          retry=len(self.brokers) - 1)
+        self.assertEqual(len(self.brokers), self.info['attempt'])
+
+    def test_ensure_four_retry(self):
+        mock_callback = mock.Mock(side_effect=IOError)
+        self.assertRaises(messaging.MessageDeliveryFailure,
+                          self.connection.ensure, None, mock_callback,
+                          retry=4)
+        self.assertEqual(5, self.info['attempt'])
+        self.assertEqual(1, mock_callback.call_count)
+
+    def test_ensure_one_retry(self):
+        mock_callback = mock.Mock(side_effect=IOError)
+        self.assertRaises(messaging.MessageDeliveryFailure,
+                          self.connection.ensure, None, mock_callback,
+                          retry=1)
+        self.assertEqual(2, self.info['attempt'])
+        self.assertEqual(1, mock_callback.call_count)
+
+    def test_ensure_no_retry(self):
+        mock_callback = mock.Mock(side_effect=IOError)
+        self.assertRaises(messaging.MessageDeliveryFailure,
+                          self.connection.ensure, None, mock_callback,
+                          retry=0)
+        self.assertEqual(1, self.info['attempt'])
+        self.assertEqual(1, mock_callback.call_count)
 
=== added directory '.pc/0005-Fix-possible-usage-of-undefined-variable.patch'
=== added file '.pc/0005-Fix-possible-usage-of-undefined-variable.patch/.timestamp'
=== added directory '.pc/0005-Fix-possible-usage-of-undefined-variable.patch/oslo'
=== added directory '.pc/0005-Fix-possible-usage-of-undefined-variable.patch/oslo/messaging'
=== added directory '.pc/0005-Fix-possible-usage-of-undefined-variable.patch/oslo/messaging/_drivers'
=== added file '.pc/0005-Fix-possible-usage-of-undefined-variable.patch/oslo/messaging/_drivers/impl_rabbit.py'
--- .pc/0005-Fix-possible-usage-of-undefined-variable.patch/oslo/messaging/_drivers/impl_rabbit.py	1970-01-01 00:00:00 +0000
+++ .pc/0005-Fix-possible-usage-of-undefined-variable.patch/oslo/messaging/_drivers/impl_rabbit.py	2015-12-17 11:03:34 +0000
@@ -0,0 +1,839 @@
+#    Copyright 2011 OpenStack Foundation
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+import itertools
+import logging
+import random
+import socket
+import ssl
+import time
+import uuid
+
+import kombu
+import kombu.connection
+import kombu.entity
+import kombu.messaging
+import six
+
+from oslo.config import cfg
+from oslo.messaging._drivers import amqp as rpc_amqp
+from oslo.messaging._drivers import amqpdriver
+from oslo.messaging._drivers import common as rpc_common
+from oslo.messaging import exceptions
+from oslo.messaging.openstack.common.gettextutils import _
+from oslo.utils import netutils
+
+rabbit_opts = [
+    cfg.StrOpt('kombu_ssl_version',
+               default='',
+               help='SSL version to use (valid only if SSL enabled). '
+                    'valid values are TLSv1, SSLv23 and SSLv3. SSLv2 may '
+                    'be available on some distributions.'
+               ),
+    cfg.StrOpt('kombu_ssl_keyfile',
+               default='',
+               help='SSL key file (valid only if SSL enabled).'),
+    cfg.StrOpt('kombu_ssl_certfile',
+               default='',
+               help='SSL cert file (valid only if SSL enabled).'),
+    cfg.StrOpt('kombu_ssl_ca_certs',
+               default='',
+               help='SSL certification authority file '
+                    '(valid only if SSL enabled).'),
+    cfg.FloatOpt('kombu_reconnect_delay',
+                 default=1.0,
+                 help='How long to wait before reconnecting in response to an '
+                      'AMQP consumer cancel notification.'),
+    cfg.StrOpt('rabbit_host',
+               default='localhost',
+               help='The RabbitMQ broker address where a single node is '
+                    'used.'),
+    cfg.IntOpt('rabbit_port',
+               default=5672,
+               help='The RabbitMQ broker port where a single node is used.'),
+    cfg.ListOpt('rabbit_hosts',
+                default=['$rabbit_host:$rabbit_port'],
+                help='RabbitMQ HA cluster host:port pairs.'),
+    cfg.BoolOpt('rabbit_use_ssl',
+                default=False,
+                help='Connect over SSL for RabbitMQ.'),
+    cfg.StrOpt('rabbit_userid',
+               default='guest',
+               help='The RabbitMQ userid.'),
+    cfg.StrOpt('rabbit_password',
+               default='guest',
+               help='The RabbitMQ password.',
+               secret=True),
+    cfg.StrOpt('rabbit_login_method',
+               default='AMQPLAIN',
+               help='the RabbitMQ login method'),
+    cfg.StrOpt('rabbit_virtual_host',
+               default='/',
+               help='The RabbitMQ virtual host.'),
+    cfg.IntOpt('rabbit_retry_interval',
+               default=1,
+               help='How frequently to retry connecting with RabbitMQ.'),
+    cfg.IntOpt('rabbit_retry_backoff',
+               default=2,
+               help='How long to backoff for between retries when connecting '
+                    'to RabbitMQ.'),
+    cfg.IntOpt('rabbit_max_retries',
+               default=0,
+               help='Maximum number of RabbitMQ connection retries. '
+                    'Default is 0 (infinite retry count).'),
+    cfg.BoolOpt('rabbit_ha_queues',
+                default=False,
+                help='Use HA queues in RabbitMQ (x-ha-policy: all). '
+                     'If you change this option, you must wipe the '
+                     'RabbitMQ database.'),
+
+    # FIXME(markmc): this was toplevel in openstack.common.rpc
+    cfg.BoolOpt('fake_rabbit',
+                default=False,
+                help='If passed, use a fake RabbitMQ provider.'),
+]
+
+LOG = logging.getLogger(__name__)
+
+
+def _get_queue_arguments(conf):
+    """Construct the arguments for declaring a queue.
+
+    If the rabbit_ha_queues option is set, we declare a mirrored queue
+    as described here:
+
+      http://www.rabbitmq.com/ha.html
+
+    Setting x-ha-policy to all means that the queue will be mirrored
+    to all nodes in the cluster.
+    """
+    return {'x-ha-policy': 'all'} if conf.rabbit_ha_queues else {}
+
+
+class RabbitMessage(dict):
+    def __init__(self, raw_message):
+        super(RabbitMessage, self).__init__(
+            rpc_common.deserialize_msg(raw_message.payload))
+        self._raw_message = raw_message
+
+    def acknowledge(self):
+        self._raw_message.ack()
+
+    def requeue(self):
+        self._raw_message.requeue()
+
+
+class ConsumerBase(object):
+    """Consumer base class."""
+
+    def __init__(self, channel, callback, tag, **kwargs):
+        """Declare a queue on an amqp channel.
+
+        'channel' is the amqp channel to use
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        queue name, exchange name, and other kombu options are
+        passed in here as a dictionary.
+        """
+        self.callback = callback
+        self.tag = six.text_type(tag)
+        self.kwargs = kwargs
+        self.queue = None
+        self.reconnect(channel)
+
+    def reconnect(self, channel):
+        """Re-declare the queue after a rabbit reconnect."""
+        self.channel = channel
+        self.kwargs['channel'] = channel
+        self.queue = kombu.entity.Queue(**self.kwargs)
+        self.queue.declare()
+
+    def _callback_handler(self, message, callback):
+        """Call callback with deserialized message.
+
+        Messages that are processed and ack'ed.
+        """
+
+        try:
+            callback(RabbitMessage(message))
+        except Exception:
+            LOG.exception(_("Failed to process message"
+                            " ... skipping it."))
+            message.ack()
+
+    def consume(self, *args, **kwargs):
+        """Actually declare the consumer on the amqp channel.  This will
+        start the flow of messages from the queue.  Using the
+        Connection.iterconsume() iterator will process the messages,
+        calling the appropriate callback.
+
+        If a callback is specified in kwargs, use that.  Otherwise,
+        use the callback passed during __init__()
+
+        If kwargs['nowait'] is True, then this call will block until
+        a message is read.
+
+        """
+
+        options = {'consumer_tag': self.tag}
+        options['nowait'] = kwargs.get('nowait', False)
+        callback = kwargs.get('callback', self.callback)
+        if not callback:
+            raise ValueError("No callback defined")
+
+        def _callback(raw_message):
+            message = self.channel.message_to_python(raw_message)
+            self._callback_handler(message, callback)
+
+        self.queue.consume(*args, callback=_callback, **options)
+
+    def cancel(self):
+        """Cancel the consuming from the queue, if it has started."""
+        try:
+            self.queue.cancel(self.tag)
+        except KeyError as e:
+            # NOTE(comstud): Kludge to get around a amqplib bug
+            if six.text_type(e) != "u'%s'" % self.tag:
+                raise
+        self.queue = None
+
+
+class DirectConsumer(ConsumerBase):
+    """Queue/consumer class for 'direct'."""
+
+    def __init__(self, conf, channel, msg_id, callback, tag, **kwargs):
+        """Init a 'direct' queue.
+
+        'channel' is the amqp channel to use
+        'msg_id' is the msg_id to listen on
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        Other kombu options may be passed
+        """
+        # Default options
+        options = {'durable': False,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=msg_id,
+                                         type='direct',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(DirectConsumer, self).__init__(channel,
+                                             callback,
+                                             tag,
+                                             name=msg_id,
+                                             exchange=exchange,
+                                             routing_key=msg_id,
+                                             **options)
+
+
+class TopicConsumer(ConsumerBase):
+    """Consumer class for 'topic'."""
+
+    def __init__(self, conf, channel, topic, callback, tag, exchange_name,
+                 name=None, **kwargs):
+        """Init a 'topic' queue.
+
+        :param channel: the amqp channel to use
+        :param topic: the topic to listen on
+        :paramtype topic: str
+        :param callback: the callback to call when messages are received
+        :param tag: a unique ID for the consumer on the channel
+        :param exchange_name: the exchange name to use
+        :param name: optional queue name, defaults to topic
+        :paramtype name: str
+
+        Other kombu options may be passed as keyword arguments
+        """
+        # Default options
+        options = {'durable': conf.amqp_durable_queues,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': conf.amqp_auto_delete,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=exchange_name,
+                                         type='topic',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(TopicConsumer, self).__init__(channel,
+                                            callback,
+                                            tag,
+                                            name=name or topic,
+                                            exchange=exchange,
+                                            routing_key=topic,
+                                            **options)
+
+
+class FanoutConsumer(ConsumerBase):
+    """Consumer class for 'fanout'."""
+
+    def __init__(self, conf, channel, topic, callback, tag, **kwargs):
+        """Init a 'fanout' queue.
+
+        'channel' is the amqp channel to use
+        'topic' is the topic to listen on
+        'callback' is the callback to call when messages are received
+        'tag' is a unique ID for the consumer on the channel
+
+        Other kombu options may be passed
+        """
+        unique = uuid.uuid4().hex
+        exchange_name = '%s_fanout' % topic
+        queue_name = '%s_fanout_%s' % (topic, unique)
+
+        # Default options
+        options = {'durable': False,
+                   'queue_arguments': _get_queue_arguments(conf),
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(name=exchange_name, type='fanout',
+                                         durable=options['durable'],
+                                         auto_delete=options['auto_delete'])
+        super(FanoutConsumer, self).__init__(channel, callback, tag,
+                                             name=queue_name,
+                                             exchange=exchange,
+                                             routing_key=topic,
+                                             **options)
+
+
+class Publisher(object):
+    """Base Publisher class."""
+
+    def __init__(self, channel, exchange_name, routing_key, **kwargs):
+        """Init the Publisher class with the exchange_name, routing_key,
+        and other options
+        """
+        self.exchange_name = exchange_name
+        self.routing_key = routing_key
+        self.kwargs = kwargs
+        self.reconnect(channel)
+
+    def reconnect(self, channel):
+        """Re-establish the Producer after a rabbit reconnection."""
+        self.exchange = kombu.entity.Exchange(name=self.exchange_name,
+                                              **self.kwargs)
+        self.producer = kombu.messaging.Producer(exchange=self.exchange,
+                                                 channel=channel,
+                                                 routing_key=self.routing_key)
+
+    def send(self, msg, timeout=None):
+        """Send a message."""
+        if timeout:
+            #
+            # AMQP TTL is in milliseconds when set in the header.
+            #
+            self.producer.publish(msg, headers={'ttl': (timeout * 1000)})
+        else:
+            self.producer.publish(msg)
+
+
+class DirectPublisher(Publisher):
+    """Publisher class for 'direct'."""
+    def __init__(self, conf, channel, topic, **kwargs):
+        """Init a 'direct' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+
+        options = {'durable': False,
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(DirectPublisher, self).__init__(channel, topic, topic,
+                                              type='direct', **options)
+
+
+class TopicPublisher(Publisher):
+    """Publisher class for 'topic'."""
+    def __init__(self, conf, channel, exchange_name, topic, **kwargs):
+        """Init a 'topic' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+        options = {'durable': conf.amqp_durable_queues,
+                   'auto_delete': conf.amqp_auto_delete,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(TopicPublisher, self).__init__(channel,
+                                             exchange_name,
+                                             topic,
+                                             type='topic',
+                                             **options)
+
+
+class FanoutPublisher(Publisher):
+    """Publisher class for 'fanout'."""
+    def __init__(self, conf, channel, topic, **kwargs):
+        """Init a 'fanout' publisher.
+
+        Kombu options may be passed as keyword args to override defaults
+        """
+        options = {'durable': False,
+                   'auto_delete': True,
+                   'exclusive': False}
+        options.update(kwargs)
+        super(FanoutPublisher, self).__init__(channel, '%s_fanout' % topic,
+                                              None, type='fanout', **options)
+
+
+class NotifyPublisher(TopicPublisher):
+    """Publisher class for 'notify'."""
+
+    def __init__(self, conf, channel, exchange_name, topic, **kwargs):
+        self.durable = kwargs.pop('durable', conf.amqp_durable_queues)
+        self.queue_arguments = _get_queue_arguments(conf)
+        super(NotifyPublisher, self).__init__(conf, channel, exchange_name,
+                                              topic, **kwargs)
+
+    def reconnect(self, channel):
+        super(NotifyPublisher, self).reconnect(channel)
+
+        # NOTE(jerdfelt): Normally the consumer would create the queue, but
+        # we do this to ensure that messages don't get dropped if the
+        # consumer is started after we do
+        queue = kombu.entity.Queue(channel=channel,
+                                   exchange=self.exchange,
+                                   durable=self.durable,
+                                   name=self.routing_key,
+                                   routing_key=self.routing_key,
+                                   queue_arguments=self.queue_arguments)
+        queue.declare()
+
+
+class Connection(object):
+    """Connection object."""
+
+    pools = {}
+
+    def __init__(self, conf, url):
+        self.consumers = []
+        self.conf = conf
+        self.max_retries = self.conf.rabbit_max_retries
+        # Try forever?
+        if self.max_retries <= 0:
+            self.max_retries = None
+        self.interval_start = self.conf.rabbit_retry_interval
+        self.interval_stepping = self.conf.rabbit_retry_backoff
+        # max retry-interval = 30 seconds
+        self.interval_max = 30
+        self.memory_transport = False
+
+        ssl_params = self._fetch_ssl_params()
+
+        if url.virtual_host is not None:
+            virtual_host = url.virtual_host
+        else:
+            virtual_host = self.conf.rabbit_virtual_host
+
+        self.brokers_params = []
+        if url.hosts:
+            for host in url.hosts:
+                params = {
+                    'hostname': host.hostname,
+                    'port': host.port or 5672,
+                    'userid': host.username or '',
+                    'password': host.password or '',
+                    'login_method': self.conf.rabbit_login_method,
+                    'virtual_host': virtual_host
+                }
+                if self.conf.fake_rabbit:
+                    params['transport'] = 'memory'
+                if self.conf.rabbit_use_ssl:
+                    params['ssl'] = ssl_params
+
+                self.brokers_params.append(params)
+        else:
+            # Old configuration format
+            for adr in self.conf.rabbit_hosts:
+                hostname, port = netutils.parse_host_port(
+                    adr, default_port=self.conf.rabbit_port)
+
+                params = {
+                    'hostname': hostname,
+                    'port': port,
+                    'userid': self.conf.rabbit_userid,
+                    'password': self.conf.rabbit_password,
+                    'login_method': self.conf.rabbit_login_method,
+                    'virtual_host': virtual_host
+                }
+
+                if self.conf.fake_rabbit:
+                    params['transport'] = 'memory'
+                if self.conf.rabbit_use_ssl:
+                    params['ssl'] = ssl_params
+
+                self.brokers_params.append(params)
+
+        random.shuffle(self.brokers_params)
+        self.brokers = itertools.cycle(self.brokers_params)
+
+        self.memory_transport = self.conf.fake_rabbit
+
+        self.connection = None
+        self.do_consume = None
+        self.reconnect()
+
+    # FIXME(markmc): use oslo sslutils when it is available as a library
+    _SSL_PROTOCOLS = {
+        "tlsv1": ssl.PROTOCOL_TLSv1,
+        "sslv23": ssl.PROTOCOL_SSLv23,
+        "sslv3": ssl.PROTOCOL_SSLv3
+    }
+
+    try:
+        _SSL_PROTOCOLS["sslv2"] = ssl.PROTOCOL_SSLv2
+    except AttributeError:
+        pass
+
+    @classmethod
+    def validate_ssl_version(cls, version):
+        key = version.lower()
+        try:
+            return cls._SSL_PROTOCOLS[key]
+        except KeyError:
+            raise RuntimeError(_("Invalid SSL version : %s") % version)
+
+    def _fetch_ssl_params(self):
+        """Handles fetching what ssl params should be used for the connection
+        (if any).
+        """
+        ssl_params = dict()
+
+        # http://docs.python.org/library/ssl.html - ssl.wrap_socket
+        if self.conf.kombu_ssl_version:
+            ssl_params['ssl_version'] = self.validate_ssl_version(
+                self.conf.kombu_ssl_version)
+        if self.conf.kombu_ssl_keyfile:
+            ssl_params['keyfile'] = self.conf.kombu_ssl_keyfile
+        if self.conf.kombu_ssl_certfile:
+            ssl_params['certfile'] = self.conf.kombu_ssl_certfile
+        if self.conf.kombu_ssl_ca_certs:
+            ssl_params['ca_certs'] = self.conf.kombu_ssl_ca_certs
+            # We might want to allow variations in the
+            # future with this?
+            ssl_params['cert_reqs'] = ssl.CERT_REQUIRED
+
+        # Return the extended behavior or just have the default behavior
+        return ssl_params or True
+
+    def _connect(self, broker):
+        """Connect to rabbit.  Re-establish any queues that may have
+        been declared before if we are reconnecting.  Exceptions should
+        be handled by the caller.
+        """
+        LOG.info(_("Connecting to AMQP server on "
+                   "%(hostname)s:%(port)d"), broker)
+        self.connection = kombu.connection.BrokerConnection(**broker)
+        self.connection_errors = self.connection.connection_errors
+        self.channel_errors = self.connection.channel_errors
+        if self.memory_transport:
+            # Kludge to speed up tests.
+            self.connection.transport.polling_interval = 0.0
+        self.do_consume = True
+        self.consumer_num = itertools.count(1)
+        self.connection.connect()
+        self.channel = self.connection.channel()
+        # work around 'memory' transport bug in 1.1.3
+        if self.memory_transport:
+            self.channel._new_queue('ae.undeliver')
+        for consumer in self.consumers:
+            consumer.reconnect(self.channel)
+        LOG.info(_('Connected to AMQP server on %(hostname)s:%(port)d'),
+                 broker)
+
+    def _disconnect(self):
+        if self.connection:
+            # XXX(nic): when reconnecting to a RabbitMQ cluster
+            # with mirrored queues in use, the attempt to release the
+            # connection can hang "indefinitely" somewhere deep down
+            # in Kombu.  Blocking the thread for a bit prior to
+            # release seems to kludge around the problem where it is
+            # otherwise reproduceable.
+            if self.conf.kombu_reconnect_delay > 0:
+                LOG.info(_("Delaying reconnect for %1.1f seconds...") %
+                         self.conf.kombu_reconnect_delay)
+                time.sleep(self.conf.kombu_reconnect_delay)
+
+            try:
+                self.connection.release()
+            except self.connection_errors: