Merge lp:~ubuntu-branches/ubuntu/trusty/oslo.messaging/trusty-updates-201602230032 into lp:ubuntu/trusty-updates/oslo.messaging

[Ubuntu Package Importer]

The package importer has detected a possible inconsistency between the package history in the archive and the history in bzr. As the archive is authoritative the importer has made lp:ubuntu/trusty-updates/oslo.messaging reflect what is in the archive and the old bzr branch has been pushed to lp:~ubuntu-branches/ubuntu/trusty/oslo.messaging/trusty-updates-201602230032. This merge proposal was created so that an Ubuntu developer can review the situations and perform a merge/upload if necessary. There are three typical cases where this can happen.
  1. Where someone pushes a change to bzr and someone else uploads the package without that change. This is the reason that this check is done by the importer. If this appears to be the case then a merge/upload should be done if the changes that were in bzr are still desirable.
  2. The importer incorrectly detected the above situation when someone made a change in bzr and then uploaded it.
  3. The importer incorrectly detected the above situation when someone just uploaded a package and didn't touch bzr.

If this case doesn't appear to be the first situation then set the status of the merge proposal to "Rejected" and help avoid the problem in future by filing a bug at https://bugs.launchpad.net/udd linking to this merge proposal.

(this is an automatically generated message)

Unmerged revisions

9. By Chuck Short on 2016-01-22

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

=== removed file '.pc/.quilt_patches'
--- .pc/.quilt_patches	2013-08-14 14:11:08 +0000
+++ .pc/.quilt_patches	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-debian/patches

=== removed file '.pc/.quilt_series'
--- .pc/.quilt_series	2013-08-14 14:11:08 +0000
+++ .pc/.quilt_series	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-series

=== removed file '.pc/.version'
--- .pc/.version	2013-08-14 14:11:08 +0000
+++ .pc/.version	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-2

=== removed directory '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch'
=== removed directory '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo'
=== removed directory '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging'
=== removed directory '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers'
=== removed file '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/amqpdriver.py'
--- .pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/amqpdriver.py	2015-04-23 15:56:08 +0000
+++ .pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/amqpdriver.py	1970-01-01 00:00:00 +0000
@@ -1,443 +0,0 @@
-
-# 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 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):
-        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):
-        while True:
-            if self.incoming:
-                return self.incoming.pop(0)
-            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' % str(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=[]):
-        super(AMQPDriverBase, self).__init__(conf, url, default_exchange,
-                                             allowed_remote_exmods)
-
-        self._server_params = self._server_params_from_url(self._url)
-
-        self._default_exchange = default_exchange
-
-        # FIXME(markmc): temp hack
-        if self._default_exchange:
-            self.conf.set_override('control_exchange', self._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 _server_params_from_url(self, url):
-        sp = {}
-
-        if url.virtual_host is not None:
-            sp['virtual_host'] = url.virtual_host
-
-        if url.hosts:
-            # FIXME(markmc): support multiple hosts
-            host = url.hosts[0]
-
-            sp['hostname'] = host.hostname
-            if host.port is not None:
-                sp['port'] = host.port
-            sp['username'] = host.username or ''
-            sp['password'] = host.password or ''
-
-        return sp
-
-    def _get_connection(self, pooled=True):
-        # FIXME(markmc): we don't yet have a connection pool for each
-        # Transport instance, so we'll only use the pool with the
-        # transport configuration from the config file
-        server_params = self._server_params or None
-        if server_params:
-            pooled = False
-        return rpc_amqp.ConnectionContext(self.conf,
-                                          self._connection_pool,
-                                          pooled=pooled,
-                                          server_params=server_params)
-
-    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):
-
-        # 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(target.topic, msg)
-                elif target.fanout:
-                    conn.fanout_send(target.topic, msg)
-                else:
-                    topic = target.topic
-                    if target.server:
-                        topic = '%s.%s' % (target.topic, target.server)
-                    conn.topic_send(topic, msg, timeout=timeout)
-
-            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):
-        return self._send(target, ctxt, message, wait_for_reply, timeout)
-
-    def send_notification(self, target, ctxt, message, version):
-        return self._send(target, ctxt, message,
-                          envelope=(version == 2.0), notify=True)
-
-    def listen(self, target):
-        conn = self._get_connection(pooled=False)
-
-        listener = AMQPListener(self, conn)
-
-        conn.declare_topic_consumer(target.topic, listener)
-        conn.declare_topic_consumer('%s.%s' % (target.topic, target.server),
-                                    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('%s.%s' % (target.topic, priority),
-                                        callback=listener,
-                                        exchange_name=target.exchange)
-        return listener
-
-    def cleanup(self):
-        if self._connection_pool:
-            self._connection_pool.empty()
-        self._connection_pool = None

=== removed file '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/common.py'
--- .pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/common.py	2015-04-23 15:56:08 +0000
+++ .pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/common.py	1970-01-01 00:00:00 +0000
@@ -1,509 +0,0 @@
-# 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
-
-from oslo.config import cfg
-from oslo import messaging
-import six
-
-from oslo.messaging import _utils as utils
-from oslo.messaging.openstack.common import importutils
-from oslo.messaging.openstack.common import jsonutils
-
-# FIXME(markmc): remove this
-_ = lambda s: s
-
-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'
-
-_exception_opts = [
-    cfg.ListOpt('allowed_rpc_exception_modules',
-                default=['oslo.messaging.exceptions',
-                         'nova.exception',
-                         'cinder.exception',
-                         _EXCEPTIONS_MODULE,
-                         ],
-                help='Modules of exceptions that are permitted to be '
-                     'recreated upon receiving exception data from an rpc '
-                     'call.'),
-]
-
-
-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 RemoteError(RPCException):
-    """Signifies that a remote class has raised an exception.
-
-    Contains a string representation of the type of the original exception,
-    the value of the original exception, and the traceback.  These are
-    sent to the parent as a joined string so printing the exception
-    contains all of the relevant info.
-
-    """
-    msg_fmt = _("Remote error: %(exc_type)s %(value)s\n%(traceback)s.")
-
-    def __init__(self, exc_type=None, value=None, traceback=None):
-        self.exc_type = exc_type
-        self.value = value
-        self.traceback = traceback
-        super(RemoteError, self).__init__(exc_type=exc_type,
-                                          value=value,
-                                          traceback=traceback)
-
-
-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 create_consumer(self, topic, proxy, fanout=False):
-        """Create a consumer on this connection.
-
-        A consumer is associated with a message queue on the backend message
-        bus.  The consumer will read messages from the queue, unpack them, and
-        dispatch them to the proxy object.  The contents of the message pulled
-        off of the queue will determine which method gets called on the proxy
-        object.
-
-        :param topic: This is a name associated with what to consume from.
-                      Multiple instances of a service may consume from the same
-                      topic. For example, all instances of nova-compute consume
-                      from a queue called "compute".  In that case, the
-                      messages will get distributed amongst the consumers in a
-                      round-robin fashion if fanout=False.  If fanout=True,
-                      every consumer associated with this topic will get a
-                      copy of every message.
-        :param proxy: The object that will handle all incoming messages.
-        :param fanout: Whether or not this is a fanout topic.  See the
-                       documentation for the topic parameter for some
-                       additional comments on this.
-        """
-        raise NotImplementedError()
-
-    def create_worker(self, topic, proxy, pool_name):
-        """Create a worker on this connection.
-
-        A worker is like a regular consumer of messages directed to a
-        topic, except that it is part of a set of such consumers (the
-        "pool") which may run in parallel. Every pool of workers will
-        receive a given message, but only one worker in the pool will
-        be asked to process it. Load is distributed across the members
-        of the pool in round-robin fashion.
-
-        :param topic: This is a name associated with what to consume from.
-                      Multiple instances of a service may consume from the same
-                      topic.
-        :param proxy: The object that will handle all incoming messages.
-        :param pool_name: String containing the name of the pool of workers
-        """
-        raise NotImplementedError()
-
-    def join_consumer_pool(self, callback, pool_name, topic, exchange_name):
-        """Register as a member of a group of consumers.
-
-        Uses given topic from the specified exchange.
-        Exactly one member of a given pool will receive each message.
-
-        A message will be delivered to multiple pools, if more than
-        one is created.
-
-        :param callback: Callable to be invoked for each message.
-        :type callback: callable accepting one argument
-        :param pool_name: The name of the consumer pool.
-        :type pool_name: str
-        :param topic: The routing topic for desired messages.
-        :type topic: str
-        :param exchange_name: The name of the message exchange where
-                              the client should attach. Defaults to
-                              the configured exchange.
-        :type exchange_name: str
-        """
-        raise NotImplementedError()
-
-    def consume_in_thread(self):
-        """Spawn a thread to handle incoming messages.
-
-        Spawn a thread that will be responsible for handling all incoming
-        messages for consumers that were set up on this connection.
-
-        Message dispatching inside of this is expected to be implemented in a
-        non-blocking manner.  An example implementation would be having this
-        thread pull messages in for all of the consumers, but utilize a thread
-        pool for dispatching the messages to the proxy objects.
-        """
-        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 = str(failure.__class__.__name__)
-    mod_name = str(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(str(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:
-        mod = importutils.import_module(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
-
-    def elevated(self, read_deleted=None, overwrite=False):
-        """Return a version of this context with admin flag set."""
-        # TODO(russellb) This method is a bit of a nova-ism.  It makes
-        # some assumptions about the data in the request context sent
-        # across rpc, while the rest of this class does not.  We could get
-        # rid of this if we changed the nova code that uses this to
-        # convert the RpcContext back to its native RequestContext doing
-        # something like nova.context.RequestContext.from_dict(ctxt.to_dict())
-
-        context = self.deepcopy()
-        context.values['is_admin'] = True
-
-        context.values.setdefault('roles', [])
-
-        if 'admin' not in context.values['roles']:
-            context.values['roles'].append('admin')
-
-        if read_deleted is not None:
-            context.values['read_deleted'] = read_deleted
-
-        return context
-
-
-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 catch_client_exception(exceptions, func, *args, **kwargs):
-    try:
-        return func(*args, **kwargs)
-    except Exception as e:
-        if type(e) in exceptions:
-            raise ClientException()
-        else:
-            raise
-
-
-def client_exceptions(*exceptions):
-    """Decorator for manager methods that raise expected exceptions.
-
-    Marking a Manager method with this decorator allows the declaration
-    of expected exceptions that the RPC layer should not consider fatal,
-    and not log as if they were generated in a real error scenario. Note
-    that this will cause listed exceptions to be wrapped in a
-    ClientException, which is used internally by the RPC layer.
-    """
-    def outer(func):
-        def inner(*args, **kwargs):
-            return catch_client_exception(exceptions, func, *args, **kwargs)
-        return inner
-    return outer
-
-
-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

=== removed file '.pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/impl_rabbit.py'
--- .pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/impl_rabbit.py	2015-04-23 15:56:08 +0000
+++ .pc/0001-rabbit-more-precise-iterconsume-timeout.patch/oslo/messaging/_drivers/impl_rabbit.py	1970-01-01 00:00:00 +0000
@@ -1,784 +0,0 @@
-#    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 socket
-import ssl
-import time
-import uuid
-
-import kombu
-import kombu.connection
-import kombu.entity
-import kombu.messaging
-from oslo.config import cfg
-import six
-
-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.openstack.common import network_utils
-
-# FIXME(markmc): remove this
-_ = lambda s: s
-
-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 = str(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 str(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, name=None,
-                 exchange_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 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_name = exchange_name or rpc_amqp.get_control_exchange(conf)
-        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, msg_id, **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, msg_id, msg_id,
-                                              type='direct', **options)
-
-
-class TopicPublisher(Publisher):
-    """Publisher class for 'topic'."""
-    def __init__(self, conf, channel, 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)
-        exchange_name = rpc_amqp.get_control_exchange(conf)
-        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, topic, **kwargs):
-        self.durable = kwargs.pop('durable', conf.amqp_durable_queues)
-        self.queue_arguments = _get_queue_arguments(conf)
-        super(NotifyPublisher, self).__init__(conf, channel, 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."""
-
-    pool = None
-
-    def __init__(self, conf, server_params=None):
-        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
-
-        if server_params is None:
-            server_params = {}
-        # Keys to translate from server_params to kombu params
-        server_params_to_kombu_params = {'username': 'userid'}
-
-        ssl_params = self._fetch_ssl_params()
-        params_list = []
-        for adr in self.conf.rabbit_hosts:
-            hostname, port = network_utils.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': self.conf.rabbit_virtual_host,
-            }
-
-            for sp_key, value in six.iteritems(server_params):
-                p_key = server_params_to_kombu_params.get(sp_key, sp_key)
-                params[p_key] = value
-
-            if self.conf.fake_rabbit:
-                params['transport'] = 'memory'
-            if self.conf.rabbit_use_ssl:
-                params['ssl'] = ssl_params
-
-            params_list.append(params)
-
-        self.params_list = itertools.cycle(params_list)
-
-        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, params):
-        """Connect to rabbit.  Re-establish any queues that may have
-        been declared before if we are reconnecting.  Exceptions should
-        be handled by the caller.
-        """
-        if self.connection:
-            LOG.info(_("Reconnecting to AMQP server on "
-                     "%(hostname)s:%(port)d") % params)
-            try:
-                # 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)
-
-                self.connection.release()
-            except self.connection_errors:
-                pass
-            # Setting this in case the next statement fails, though
-            # it shouldn't be doing any network operations, yet.
-            self.connection = None
-        self.connection = kombu.connection.BrokerConnection(**params)
-        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') %
-                 params)
-
-    def reconnect(self):
-        """Handles reconnecting and re-establishing queues.
-        Will retry up to self.max_retries number of times.
-        self.max_retries = 0 means to retry forever.
-        Sleep between tries, starting at self.interval_start
-        seconds, backing off self.interval_stepping number of seconds
-        each attempt.
-        """
-
-        attempt = 0
-        while True:
-            params = six.next(self.params_list)
-            attempt += 1
-            try:
-                self._connect(params)
-                return
-            except IOError as e:
-                pass
-            except self.connection_errors as e:
-                pass
-            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 str(e):
-                    raise
-
-            log_info = {}
-            log_info['err_str'] = str(e)
-            log_info['max_retries'] = self.max_retries
-            log_info.update(params)
-
-            if self.max_retries and attempt == self.max_retries:
-                msg = _('Unable to connect to AMQP server on '
-                        '%(hostname)s:%(port)d after %(max_retries)d '
-                        'tries: %(err_str)s') % log_info
-                LOG.error(msg)
-                raise rpc_common.RPCException(msg)
-
-            if attempt == 1:
-                sleep_time = self.interval_start or 1
-            elif attempt > 1:
-                sleep_time += self.interval_stepping
-            if self.interval_max:
-                sleep_time = min(sleep_time, self.interval_max)
-
-            log_info['sleep_time'] = sleep_time
-            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, *args, **kwargs):
-        while True:
-            try:
-                return method(*args, **kwargs)
-            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 str(e):
-                    raise
-                if error_callback:
-                    error_callback(e)
-            self.reconnect()
-
-    def get_channel(self):
-        """Convenience call for bin/clear_rabbit_queues."""
-        return self.channel
-
-    def close(self):
-        """Close/release this 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': 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') %
-                          str(exc))
-                raise rpc_common.Timeout()
-            else:
-                LOG.exception(_('Failed to consume message from queue: %s') %
-                              str(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, **kwargs):
-        """Send to a publisher based on the publisher class."""
-
-        def _error_callback(exc):
-            log_info = {'topic': topic, 'err_str': 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, **kwargs)
-            publisher.send(msg, timeout)
-
-        self.ensure(_error_callback, _publish)
-
-    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, topic, callback=None, queue_name=None,
-                               exchange_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, topic, msg, timeout=None):
-        """Send a 'topic' message."""
-        self.publisher_send(TopicPublisher, topic, msg, timeout)
-
-    def fanout_send(self, topic, msg):
-        """Send a 'fanout' message."""
-        self.publisher_send(FanoutPublisher, topic, msg)
-
-    def notify_send(self, topic, msg, **kwargs):
-        """Send a notify message on a topic."""
-        self.publisher_send(NotifyPublisher, topic, msg, None, **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=[]):
-        conf.register_opts(rabbit_opts)
-        conf.register_opts(rpc_amqp.amqp_opts)
-
-        connection_pool = rpc_amqp.get_connection_pool(conf, Connection)
-
-        super(RabbitDriver, self).__init__(conf, url,
-                                           connection_pool,
-                                           default_exchange,
-                                           allowed_remote_exmods)
-
-    def require_features(self, requeue=True):
-        pass

=== removed directory '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch'
=== removed directory '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/oslo'
=== removed directory '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/oslo/messaging'
=== removed directory '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/oslo/messaging/_drivers'
=== removed file '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/oslo/messaging/_drivers/common.py'
--- .pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/oslo/messaging/_drivers/common.py	2015-04-23 15:56:08 +0000
+++ .pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/oslo/messaging/_drivers/common.py	1970-01-01 00:00:00 +0000
@@ -1,535 +0,0 @@
-# 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
-
-from oslo.config import cfg
-from oslo import messaging
-import six
-
-from oslo.messaging import _utils as utils
-from oslo.messaging.openstack.common import importutils
-from oslo.messaging.openstack.common import jsonutils
-
-# FIXME(markmc): remove this
-_ = lambda s: s
-
-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'
-
-_exception_opts = [
-    cfg.ListOpt('allowed_rpc_exception_modules',
-                default=['oslo.messaging.exceptions',
-                         'nova.exception',
-                         'cinder.exception',
-                         _EXCEPTIONS_MODULE,
-                         ],
-                help='Modules of exceptions that are permitted to be '
-                     'recreated upon receiving exception data from an rpc '
-                     'call.'),
-]
-
-
-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 RemoteError(RPCException):
-    """Signifies that a remote class has raised an exception.
-
-    Contains a string representation of the type of the original exception,
-    the value of the original exception, and the traceback.  These are
-    sent to the parent as a joined string so printing the exception
-    contains all of the relevant info.
-
-    """
-    msg_fmt = _("Remote error: %(exc_type)s %(value)s\n%(traceback)s.")
-
-    def __init__(self, exc_type=None, value=None, traceback=None):
-        self.exc_type = exc_type
-        self.value = value
-        self.traceback = traceback
-        super(RemoteError, self).__init__(exc_type=exc_type,
-                                          value=value,
-                                          traceback=traceback)
-
-
-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 create_consumer(self, topic, proxy, fanout=False):
-        """Create a consumer on this connection.
-
-        A consumer is associated with a message queue on the backend message
-        bus.  The consumer will read messages from the queue, unpack them, and
-        dispatch them to the proxy object.  The contents of the message pulled
-        off of the queue will determine which method gets called on the proxy
-        object.
-
-        :param topic: This is a name associated with what to consume from.
-                      Multiple instances of a service may consume from the same
-                      topic. For example, all instances of nova-compute consume
-                      from a queue called "compute".  In that case, the
-                      messages will get distributed amongst the consumers in a
-                      round-robin fashion if fanout=False.  If fanout=True,
-                      every consumer associated with this topic will get a
-                      copy of every message.
-        :param proxy: The object that will handle all incoming messages.
-        :param fanout: Whether or not this is a fanout topic.  See the
-                       documentation for the topic parameter for some
-                       additional comments on this.
-        """
-        raise NotImplementedError()
-
-    def create_worker(self, topic, proxy, pool_name):
-        """Create a worker on this connection.
-
-        A worker is like a regular consumer of messages directed to a
-        topic, except that it is part of a set of such consumers (the
-        "pool") which may run in parallel. Every pool of workers will
-        receive a given message, but only one worker in the pool will
-        be asked to process it. Load is distributed across the members
-        of the pool in round-robin fashion.
-
-        :param topic: This is a name associated with what to consume from.
-                      Multiple instances of a service may consume from the same
-                      topic.
-        :param proxy: The object that will handle all incoming messages.
-        :param pool_name: String containing the name of the pool of workers
-        """
-        raise NotImplementedError()
-
-    def join_consumer_pool(self, callback, pool_name, topic, exchange_name):
-        """Register as a member of a group of consumers.
-
-        Uses given topic from the specified exchange.
-        Exactly one member of a given pool will receive each message.
-
-        A message will be delivered to multiple pools, if more than
-        one is created.
-
-        :param callback: Callable to be invoked for each message.
-        :type callback: callable accepting one argument
-        :param pool_name: The name of the consumer pool.
-        :type pool_name: str
-        :param topic: The routing topic for desired messages.
-        :type topic: str
-        :param exchange_name: The name of the message exchange where
-                              the client should attach. Defaults to
-                              the configured exchange.
-        :type exchange_name: str
-        """
-        raise NotImplementedError()
-
-    def consume_in_thread(self):
-        """Spawn a thread to handle incoming messages.
-
-        Spawn a thread that will be responsible for handling all incoming
-        messages for consumers that were set up on this connection.
-
-        Message dispatching inside of this is expected to be implemented in a
-        non-blocking manner.  An example implementation would be having this
-        thread pull messages in for all of the consumers, but utilize a thread
-        pool for dispatching the messages to the proxy objects.
-        """
-        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 = str(failure.__class__.__name__)
-    mod_name = str(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(str(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:
-        mod = importutils.import_module(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
-
-    def elevated(self, read_deleted=None, overwrite=False):
-        """Return a version of this context with admin flag set."""
-        # TODO(russellb) This method is a bit of a nova-ism.  It makes
-        # some assumptions about the data in the request context sent
-        # across rpc, while the rest of this class does not.  We could get
-        # rid of this if we changed the nova code that uses this to
-        # convert the RpcContext back to its native RequestContext doing
-        # something like nova.context.RequestContext.from_dict(ctxt.to_dict())
-
-        context = self.deepcopy()
-        context.values['is_admin'] = True
-
-        context.values.setdefault('roles', [])
-
-        if 'admin' not in context.values['roles']:
-            context.values['roles'].append('admin')
-
-        if read_deleted is not None:
-            context.values['read_deleted'] = read_deleted
-
-        return context
-
-
-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 catch_client_exception(exceptions, func, *args, **kwargs):
-    try:
-        return func(*args, **kwargs)
-    except Exception as e:
-        if type(e) in exceptions:
-            raise ClientException()
-        else:
-            raise
-
-
-def client_exceptions(*exceptions):
-    """Decorator for manager methods that raise expected exceptions.
-
-    Marking a Manager method with this decorator allows the declaration
-    of expected exceptions that the RPC layer should not consider fatal,
-    and not log as if they were generated in a real error scenario. Note
-    that this will cause listed exceptions to be wrapped in a
-    ClientException, which is used internally by the RPC layer.
-    """
-    def outer(func):
-        def inner(*args, **kwargs):
-            return catch_client_exception(exceptions, func, *args, **kwargs)
-        return inner
-    return outer
-
-
-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)
-        return self
-
-    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)

=== removed directory '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/tests'
=== removed file '.pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/tests/test_utils.py'
--- .pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/tests/test_utils.py	2015-04-23 15:56:08 +0000
+++ .pc/0002-rabbit-fix-timeout-timer-when-duration-is-None.patch/tests/test_utils.py	1970-01-01 00:00:00 +0000
@@ -1,49 +0,0 @@
-
-# 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'))

=== removed directory '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch'
=== removed directory '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo'
=== removed directory '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging'
=== removed directory '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers'
=== removed file '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers/common.py'
--- .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers/common.py	2015-04-23 15:56:08 +0000
+++ .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers/common.py	1970-01-01 00:00:00 +0000
@@ -1,535 +0,0 @@
-# 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
-
-from oslo.config import cfg
-from oslo import messaging
-import six
-
-from oslo.messaging import _utils as utils
-from oslo.messaging.openstack.common import importutils
-from oslo.messaging.openstack.common import jsonutils
-
-# FIXME(markmc): remove this
-_ = lambda s: s
-
-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'
-
-_exception_opts = [
-    cfg.ListOpt('allowed_rpc_exception_modules',
-                default=['oslo.messaging.exceptions',
-                         'nova.exception',
-                         'cinder.exception',
-                         _EXCEPTIONS_MODULE,
-                         ],
-                help='Modules of exceptions that are permitted to be '
-                     'recreated upon receiving exception data from an rpc '
-                     'call.'),
-]
-
-
-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 RemoteError(RPCException):
-    """Signifies that a remote class has raised an exception.
-
-    Contains a string representation of the type of the original exception,
-    the value of the original exception, and the traceback.  These are
-    sent to the parent as a joined string so printing the exception
-    contains all of the relevant info.
-
-    """
-    msg_fmt = _("Remote error: %(exc_type)s %(value)s\n%(traceback)s.")
-
-    def __init__(self, exc_type=None, value=None, traceback=None):
-        self.exc_type = exc_type
-        self.value = value
-        self.traceback = traceback
-        super(RemoteError, self).__init__(exc_type=exc_type,
-                                          value=value,
-                                          traceback=traceback)
-
-
-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 create_consumer(self, topic, proxy, fanout=False):
-        """Create a consumer on this connection.
-
-        A consumer is associated with a message queue on the backend message
-        bus.  The consumer will read messages from the queue, unpack them, and
-        dispatch them to the proxy object.  The contents of the message pulled
-        off of the queue will determine which method gets called on the proxy
-        object.
-
-        :param topic: This is a name associated with what to consume from.
-                      Multiple instances of a service may consume from the same
-                      topic. For example, all instances of nova-compute consume
-                      from a queue called "compute".  In that case, the
-                      messages will get distributed amongst the consumers in a
-                      round-robin fashion if fanout=False.  If fanout=True,
-                      every consumer associated with this topic will get a
-                      copy of every message.
-        :param proxy: The object that will handle all incoming messages.
-        :param fanout: Whether or not this is a fanout topic.  See the
-                       documentation for the topic parameter for some
-                       additional comments on this.
-        """
-        raise NotImplementedError()
-
-    def create_worker(self, topic, proxy, pool_name):
-        """Create a worker on this connection.
-
-        A worker is like a regular consumer of messages directed to a
-        topic, except that it is part of a set of such consumers (the
-        "pool") which may run in parallel. Every pool of workers will
-        receive a given message, but only one worker in the pool will
-        be asked to process it. Load is distributed across the members
-        of the pool in round-robin fashion.
-
-        :param topic: This is a name associated with what to consume from.
-                      Multiple instances of a service may consume from the same
-                      topic.
-        :param proxy: The object that will handle all incoming messages.
-        :param pool_name: String containing the name of the pool of workers
-        """
-        raise NotImplementedError()
-
-    def join_consumer_pool(self, callback, pool_name, topic, exchange_name):
-        """Register as a member of a group of consumers.
-
-        Uses given topic from the specified exchange.
-        Exactly one member of a given pool will receive each message.
-
-        A message will be delivered to multiple pools, if more than
-        one is created.
-
-        :param callback: Callable to be invoked for each message.
-        :type callback: callable accepting one argument
-        :param pool_name: The name of the consumer pool.
-        :type pool_name: str
-        :param topic: The routing topic for desired messages.
-        :type topic: str
-        :param exchange_name: The name of the message exchange where
-                              the client should attach. Defaults to
-                              the configured exchange.
-        :type exchange_name: str
-        """
-        raise NotImplementedError()
-
-    def consume_in_thread(self):
-        """Spawn a thread to handle incoming messages.
-
-        Spawn a thread that will be responsible for handling all incoming
-        messages for consumers that were set up on this connection.
-
-        Message dispatching inside of this is expected to be implemented in a
-        non-blocking manner.  An example implementation would be having this
-        thread pull messages in for all of the consumers, but utilize a thread
-        pool for dispatching the messages to the proxy objects.
-        """
-        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 = str(failure.__class__.__name__)
-    mod_name = str(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(str(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:
-        mod = importutils.import_module(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
-
-    def elevated(self, read_deleted=None, overwrite=False):
-        """Return a version of this context with admin flag set."""
-        # TODO(russellb) This method is a bit of a nova-ism.  It makes
-        # some assumptions about the data in the request context sent
-        # across rpc, while the rest of this class does not.  We could get
-        # rid of this if we changed the nova code that uses this to
-        # convert the RpcContext back to its native RequestContext doing
-        # something like nova.context.RequestContext.from_dict(ctxt.to_dict())
-
-        context = self.deepcopy()
-        context.values['is_admin'] = True
-
-        context.values.setdefault('roles', [])
-
-        if 'admin' not in context.values['roles']:
-            context.values['roles'].append('admin')
-
-        if read_deleted is not None:
-            context.values['read_deleted'] = read_deleted
-
-        return context
-
-
-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 catch_client_exception(exceptions, func, *args, **kwargs):
-    try:
-        return func(*args, **kwargs)
-    except Exception as e:
-        if type(e) in exceptions:
-            raise ClientException()
-        else:
-            raise
-
-
-def client_exceptions(*exceptions):
-    """Decorator for manager methods that raise expected exceptions.
-
-    Marking a Manager method with this decorator allows the declaration
-    of expected exceptions that the RPC layer should not consider fatal,
-    and not log as if they were generated in a real error scenario. Note
-    that this will cause listed exceptions to be wrapped in a
-    ClientException, which is used internally by the RPC layer.
-    """
-    def outer(func):
-        def inner(*args, **kwargs):
-            return catch_client_exception(exceptions, func, *args, **kwargs)
-        return inner
-    return outer
-
-
-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)
-        return self
-
-    def check_return(self, timeout_callback, *args, **kwargs):
-        maximum = kwargs.pop('maximum', None)
-        if self._duration is None:
-            return None if maximum is None else maximum
-        if self._ends_at is None:
-            raise RuntimeError("Can not check/return a timeout from a timer"
-                               " that has not been started")
-
-        left = self._ends_at - time.time()
-        if left <= 0:
-            timeout_callback(*args, **kwargs)
-
-        return left if maximum is None else min(left, maximum)

=== removed file '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers/impl_rabbit.py'
--- .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers/impl_rabbit.py	2015-04-23 15:56:08 +0000
+++ .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/oslo/messaging/_drivers/impl_rabbit.py	1970-01-01 00:00:00 +0000
@@ -1,793 +0,0 @@
-#    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 socket
-import ssl
-import time
-import uuid
-
-import kombu
-import kombu.connection
-import kombu.entity
-import kombu.messaging
-from oslo.config import cfg
-import six
-
-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.openstack.common import network_utils
-
-# FIXME(markmc): remove this
-_ = lambda s: s
-
-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 = str(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 str(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, name=None,
-                 exchange_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 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_name = exchange_name or rpc_amqp.get_control_exchange(conf)
-        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, msg_id, **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, msg_id, msg_id,
-                                              type='direct', **options)
-
-
-class TopicPublisher(Publisher):
-    """Publisher class for 'topic'."""
-    def __init__(self, conf, channel, 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)
-        exchange_name = rpc_amqp.get_control_exchange(conf)
-        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, topic, **kwargs):
-        self.durable = kwargs.pop('durable', conf.amqp_durable_queues)
-        self.queue_arguments = _get_queue_arguments(conf)
-        super(NotifyPublisher, self).__init__(conf, channel, 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."""
-
-    pool = None
-
-    def __init__(self, conf, server_params=None):
-        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
-
-        if server_params is None:
-            server_params = {}
-        # Keys to translate from server_params to kombu params
-        server_params_to_kombu_params = {'username': 'userid'}
-
-        ssl_params = self._fetch_ssl_params()
-        params_list = []
-        for adr in self.conf.rabbit_hosts:
-            hostname, port = network_utils.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': self.conf.rabbit_virtual_host,
-            }
-
-            for sp_key, value in six.iteritems(server_params):
-                p_key = server_params_to_kombu_params.get(sp_key, sp_key)
-                params[p_key] = value
-
-            if self.conf.fake_rabbit:
-                params['transport'] = 'memory'
-            if self.conf.rabbit_use_ssl:
-                params['ssl'] = ssl_params
-
-            params_list.append(params)
-
-        self.params_list = itertools.cycle(params_list)
-
-        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, params):
-        """Connect to rabbit.  Re-establish any queues that may have
-        been declared before if we are reconnecting.  Exceptions should
-        be handled by the caller.
-        """
-        if self.connection:
-            LOG.info(_("Reconnecting to AMQP server on "
-                     "%(hostname)s:%(port)d") % params)
-            try:
-                # 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)
-
-                self.connection.release()
-            except self.connection_errors:
-                pass
-            # Setting this in case the next statement fails, though
-            # it shouldn't be doing any network operations, yet.
-            self.connection = None
-        self.connection = kombu.connection.BrokerConnection(**params)
-        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') %
-                 params)
-
-    def reconnect(self):
-        """Handles reconnecting and re-establishing queues.
-        Will retry up to self.max_retries number of times.
-        self.max_retries = 0 means to retry forever.
-        Sleep between tries, starting at self.interval_start
-        seconds, backing off self.interval_stepping number of seconds
-        each attempt.
-        """
-
-        attempt = 0
-        while True:
-            params = six.next(self.params_list)
-            attempt += 1
-            try:
-                self._connect(params)
-                return
-            except IOError as e:
-                pass
-            except self.connection_errors as e:
-                pass
-            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 str(e):
-                    raise
-
-            log_info = {}
-            log_info['err_str'] = str(e)
-            log_info['max_retries'] = self.max_retries
-            log_info.update(params)
-
-            if self.max_retries and attempt == self.max_retries:
-                msg = _('Unable to connect to AMQP server on '
-                        '%(hostname)s:%(port)d after %(max_retries)d '
-                        'tries: %(err_str)s') % log_info
-                LOG.error(msg)
-                raise rpc_common.RPCException(msg)
-
-            if attempt == 1:
-                sleep_time = self.interval_start or 1
-            elif attempt > 1:
-                sleep_time += self.interval_stepping
-            if self.interval_max:
-                sleep_time = min(sleep_time, self.interval_max)
-
-            log_info['sleep_time'] = sleep_time
-            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, *args, **kwargs):
-        while True:
-            try:
-                return method(*args, **kwargs)
-            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 str(e):
-                    raise
-                if error_callback:
-                    error_callback(e)
-            self.reconnect()
-
-    def get_channel(self):
-        """Convenience call for bin/clear_rabbit_queues."""
-        return self.channel
-
-    def close(self):
-        """Close/release this 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': 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):
-            self.do_consume = True
-            timer.check_return(_raise_timeout, exc)
-            LOG.exception(_('Failed to consume message from queue: %s'),
-                          exc)
-
-        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, **kwargs):
-        """Send to a publisher based on the publisher class."""
-
-        def _error_callback(exc):
-            log_info = {'topic': topic, 'err_str': 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, **kwargs)
-            publisher.send(msg, timeout)
-
-        self.ensure(_error_callback, _publish)
-
-    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, topic, callback=None, queue_name=None,
-                               exchange_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, topic, msg, timeout=None):
-        """Send a 'topic' message."""
-        self.publisher_send(TopicPublisher, topic, msg, timeout)
-
-    def fanout_send(self, topic, msg):
-        """Send a 'fanout' message."""
-        self.publisher_send(FanoutPublisher, topic, msg)
-
-    def notify_send(self, topic, msg, **kwargs):
-        """Send a notify message on a topic."""
-        self.publisher_send(NotifyPublisher, topic, msg, None, **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=[]):
-        conf.register_opts(rabbit_opts)
-        conf.register_opts(rpc_amqp.amqp_opts)
-
-        connection_pool = rpc_amqp.get_connection_pool(conf, Connection)
-
-        super(RabbitDriver, self).__init__(conf, url,
-                                           connection_pool,
-                                           default_exchange,
-                                           allowed_remote_exmods)
-
-    def require_features(self, requeue=True):
-        pass

=== removed directory '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests'
=== removed file '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests/test_rabbit.py'
--- .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests/test_rabbit.py	2015-04-23 15:56:08 +0000
+++ .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests/test_rabbit.py	1970-01-01 00:00:00 +0000
@@ -1,646 +0,0 @@
-# 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 sys
-import threading
-import uuid
-
-import fixtures
-import kombu
-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=None)),
-        ('empty',
-         dict(url='rabbit:///',
-              expected=dict(virtual_host=''))),
-        ('localhost',
-         dict(url='rabbit://localhost/',
-              expected=dict(hostname='localhost',
-                            username='',
-                            password='',
-                            virtual_host=''))),
-        ('virtual_host',
-         dict(url='rabbit:///vhost',
-              expected=dict(virtual_host='vhost'))),
-        ('no_creds',
-         dict(url='rabbit://host/virtual_host',
-              expected=dict(hostname='host',
-                            username='',
-                            password='',
-                            virtual_host='virtual_host'))),
-        ('no_port',
-         dict(url='rabbit://user:password@host/virtual_host',
-              expected=dict(hostname='host',
-                            username='user',
-                            password='password',
-                            virtual_host='virtual_host'))),
-        ('full_url',
-         dict(url='rabbit://user:password@host:10/virtual_host',
-              expected=dict(hostname='host',
-                            port=10,
-                            username='user',
-                            password='password',
-                            virtual_host='virtual_host'))),
-    ]
-
-    def setUp(self):
-        super(TestRabbitTransportURL, self).setUp()
-
-        self.messaging_conf.transport_driver = 'rabbit'
-        self.messaging_conf.in_memory = True
-
-        self._server_params = []
-        cnx_init = rabbit_driver.Connection.__init__
-
-        def record_params(cnx, conf, server_params=None):
-            self._server_params.append(server_params)
-            return cnx_init(cnx, conf, server_params)
-
-        def dummy_send(cnx, topic, msg, timeout=None):
-            pass
-
-        self.stubs.Set(rabbit_driver.Connection, '__init__', record_params)
-        self.stubs.Set(rabbit_driver.Connection, 'topic_send', dummy_send)
-
-        self._driver = messaging.get_transport(self.conf, self.url)._driver
-        self._target = messaging.Target(topic='testtopic')
-
-    def test_transport_url_listen(self):
-        self._driver.listen(self._target)
-        self.assertEqual(self._server_params[0], self.expected)
-
-    def test_transport_url_listen_for_notification(self):
-        self._driver.listen_for_notifications(
-            [(messaging.Target(topic='topic'), 'info')])
-        self.assertEqual(self._server_params[0], self.expected)
-
-    def test_transport_url_send(self):
-        self._driver.send(self._target, {}, {})
-        self.assertEqual(self._server_params[0], self.expected)
-
-
-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(received.ctxt, self.ctxt)
-            self.assertEqual(received.message, {'tx_id': i})
-            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(replies), len(senders))
-        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(reply, {'rx_id': order[i]})
-            else:
-                self.assertEqual(reply, self.reply)
-
-        if not self.timeout and self.failure and not self.expected:
-            self.assertTrue(len(errors) > 0, errors)
-        else:
-            self.assertEqual(len(errors), 0, errors)
-
-
-TestSendReceive.generate_scenarios()
-
-
-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:
-                with wait_conditions[0]:
-                    wait_conditions.pop().wait()
-            return orig_reply_waiter(self, msg_id, timeout)
-
-        self.stubs.Set(amqpdriver.ReplyWaiter, 'wait', reply_waiter)
-
-        def send_and_wait_for_reply(i):
-            replies.append(driver.send(target,
-                                       {},
-                                       {'tx_id': i},
-                                       wait_for_reply=True,
-                                       timeout=None))
-
-        while len(senders) < 2:
-            t = threading.Thread(target=send_and_wait_for_reply,
-                                 args=(len(senders), ))
-            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)
-        senders[0].start()
-
-        msgs.append(listener.poll())
-        self.assertEqual(msgs[-1].message, {'tx_id': 0})
-
-        # Start the second guy, receive his message
-        senders[1].start()
-
-        msgs.append(listener.poll())
-        self.assertEqual(msgs[-1].message, {'tx_id': 1})
-
-        # 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()
-
-        # 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(replies), len(senders))
-        self.assertEqual(replies[0], {'rx_id': 1})
-        self.assertEqual(replies[1], {'rx_id': 0})
-
-
-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 test_reconnect_order(self):
-        brokers = ['host1', 'host2', 'host3', 'host4', 'host5']
-        brokers_count = len(brokers)
-
-        self.conf.rabbit_hosts = brokers
-        self.conf.rabbit_max_retries = 1
-
-        info = {'attempt': 0}
-
-        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
-
-            expected_broker = brokers[info['attempt'] % brokers_count]
-            self.assertEqual(params['hostname'], expected_broker)
-
-            info['attempt'] += 1
-
-        # 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
-        connection = rabbit_driver.Connection(self.conf)
-
-        # now that we have connection object, revert to the real 'connect'
-        # implementation
-        self.stubs.UnsetAll()
-
-        for i in range(len(brokers)):
-            self.assertRaises(driver_common.RPCException, connection.reconnect)
-
-        connection.close()

=== removed file '.pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests/test_utils.py'
--- .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests/test_utils.py	2015-04-23 15:56:08 +0000
+++ .pc/0003-Declare-DirectPublisher-exchanges-with-passive-True.patch/tests/test_utils.py	1970-01-01 00:00:00 +0000
@@ -1,64 +0,0 @@
-
-# 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._drivers import common
-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'))
-
-
-class TimerTestCase(test_utils.BaseTestCase):
-    def test_duration_is_none(self):
-        t = common.DecayingTimer()
-        t.start()
-        remaining = t.check_return(None)
-        self.assertEqual(None, remaining)
-
-    def test_duration_is_none_and_maximun_set(self):
-        t = common.DecayingTimer()
-        t.start()
-        remaining = t.check_return(None, maximum=2)
-        self.assertEqual(2, remaining)

=== renamed file '.pc/applied-patches' => '.pc/applied-patches.THIS'
=== removed directory '.pc/redeclare-consumers-when-ack-requeue-fails.patch'
=== removed directory '.pc/redeclare-consumers-when-ack-requeue-fails.patch/oslo'
=== removed directory '.pc/redeclare-consumers-when-ack-requeue-fails.patch/oslo/messaging'
=== removed directory '.pc/redeclare-consumers-when-ack-requeue-fails.patch/oslo/messaging/_drivers'
=== removed file '.pc/redeclare-consumers-when-ack-requeue-fails.patch/oslo/messaging/_drivers/impl_rabbit.py'
--- .pc/redeclare-consumers-when-ack-requeue-fails.patch/oslo/messaging/_drivers/impl_rabbit.py	2015-06-25 09:59:42 +0000
+++ .pc/redeclare-consumers-when-ack-requeue-fails.patch/oslo/messaging/_drivers/impl_rabbit.py	1970-01-01 00:00:00 +0000
@@ -1,819 +0,0 @@
-#    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 socket
-import ssl
-import time
-import uuid
-
-import kombu
-import kombu.connection
-import kombu.entity
-import kombu.messaging
-from oslo.config import cfg
-import six
-
-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.openstack.common import network_utils
-
-# FIXME(markmc): remove this
-_ = lambda s: s
-
-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 = str(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 str(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, name=None,
-                 exchange_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 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_name = exchange_name or rpc_amqp.get_control_exchange(conf)
-        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, msg_id, **kwargs):
-        """Init a 'direct' publisher.
-
-        Kombu options may be passed as keyword args to override defaults
-        """
-
-        options = {'durable': False,
-                   'auto_delete': True,
-                   'exclusive': False,
-                   'passive': True}
-        options.update(kwargs)
-        super(DirectPublisher, self).__init__(channel, msg_id, msg_id,
-                                              type='direct', **options)
-
-
-class TopicPublisher(Publisher):
-    """Publisher class for 'topic'."""
-    def __init__(self, conf, channel, 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)
-        exchange_name = rpc_amqp.get_control_exchange(conf)
-        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, topic, **kwargs):
-        self.durable = kwargs.pop('durable', conf.amqp_durable_queues)
-        self.queue_arguments = _get_queue_arguments(conf)
-        super(NotifyPublisher, self).__init__(conf, channel, 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."""
-
-    pool = None
-
-    def __init__(self, conf, server_params=None):
-        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
-
-        if server_params is None:
-            server_params = {}
-        # Keys to translate from server_params to kombu params
-        server_params_to_kombu_params = {'username': 'userid'}
-
-        ssl_params = self._fetch_ssl_params()
-        params_list = []
-        for adr in self.conf.rabbit_hosts:
-            hostname, port = network_utils.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': self.conf.rabbit_virtual_host,
-            }
-
-            for sp_key, value in six.iteritems(server_params):
-                p_key = server_params_to_kombu_params.get(sp_key, sp_key)
-                params[p_key] = value
-
-            if self.conf.fake_rabbit:
-                params['transport'] = 'memory'
-            if self.conf.rabbit_use_ssl:
-                params['ssl'] = ssl_params
-
-            params_list.append(params)
-
-        self.params_list = itertools.cycle(params_list)
-
-        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, params):
-        """Connect to rabbit.  Re-establish any queues that may have
-        been declared before if we are reconnecting.  Exceptions should
-        be handled by the caller.
-        """
-        if self.connection:
-            LOG.info(_("Reconnecting to AMQP server on "
-                     "%(hostname)s:%(port)d") % params)
-            try:
-                # 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)
-
-                self.connection.release()
-            except self.connection_errors:
-                pass
-            # Setting this in case the next statement fails, though
-            # it shouldn't be doing any network operations, yet.
-            self.connection = None
-        self.connection = kombu.connection.BrokerConnection(**params)
-        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()