Add rpc to openstack-common.

Implements blueprint common-rpc.

This patch imports nova.rpc to openstack-common.  All of the necessary
changes to the core code to make it compatible with openstack-common
have been done in nova, so the only changes made here are to imports.

There are more changes made to the tests, but nothing that changes the
core functionality of the tests.

Change-Id: I17330aa4adfd0f22c449a2376833c7fe9dfd0cf1

9 files changed, 3025 insertions, 0 deletions

diff --git a/openstack/common/rpc/__init__.py b/openstack/common/rpc/__init__.py
new file mode 100644
index 0000000..116aa84
--- /dev/null
+++ b/openstack/common/rpc/__init__.py
@@ -0,0 +1,248 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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.
+
+"""
+A remote procedure call (rpc) abstraction.
+
+For some wrappers that add message versioning to rpc, see:
+    rpc.dispatcher
+    rpc.proxy
+"""
+
+from openstack.common import cfg
+from openstack.common import importutils
+
+
+rpc_opts = [
+    cfg.StrOpt('rpc_backend',
+               default='nova.rpc.impl_kombu',
+               help="The messaging module to use, defaults to kombu."),
+    cfg.IntOpt('rpc_thread_pool_size',
+               default=64,
+               help='Size of RPC thread pool'),
+    cfg.IntOpt('rpc_conn_pool_size',
+               default=30,
+               help='Size of RPC connection pool'),
+    cfg.IntOpt('rpc_response_timeout',
+               default=60,
+               help='Seconds to wait for a response from call or multicall'),
+    cfg.ListOpt('allowed_rpc_exception_modules',
+                default=['openstack.common.exception', 'nova.exception'],
+                help='Modules of exceptions that are permitted to be recreated'
+                     'upon receiving exception data from an rpc call.'),
+    cfg.StrOpt('control_exchange',
+               default='nova',
+               help='AMQP exchange to connect to if using RabbitMQ or Qpid'),
+    cfg.BoolOpt('fake_rabbit',
+                default=False,
+                help='If passed, use a fake RabbitMQ provider'),
+    ]
+
+cfg.CONF.register_opts(rpc_opts)
+
+
+def create_connection(new=True):
+    """Create a connection to the message bus used for rpc.
+
+    For some example usage of creating a connection and some consumers on that
+    connection, see nova.service.
+
+    :param new: Whether or not to create a new connection.  A new connection
+                will be created by default.  If new is False, the
+                implementation is free to return an existing connection from a
+                pool.
+
+    :returns: An instance of nova.rpc.common.Connection
+    """
+    return _get_impl().create_connection(cfg.CONF, new=new)
+
+
+def call(context, topic, msg, timeout=None):
+    """Invoke a remote method that returns something.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param topic: The topic to send the rpc message to.  This correlates to the
+                  topic argument of
+                  nova.rpc.common.Connection.create_consumer() and only applies
+                  when the consumer was created with fanout=False.
+    :param msg: This is a dict in the form { "method" : "method_to_invoke",
+                                             "args" : dict_of_kwargs }
+    :param timeout: int, number of seconds to use for a response timeout.
+                    If set, this overrides the rpc_response_timeout option.
+
+    :returns: A dict from the remote method.
+
+    :raises: nova.rpc.common.Timeout if a complete response is not received
+             before the timeout is reached.
+    """
+    return _get_impl().call(cfg.CONF, context, topic, msg, timeout)
+
+
+def cast(context, topic, msg):
+    """Invoke a remote method that does not return anything.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param topic: The topic to send the rpc message to.  This correlates to the
+                  topic argument of
+                  nova.rpc.common.Connection.create_consumer() and only applies
+                  when the consumer was created with fanout=False.
+    :param msg: This is a dict in the form { "method" : "method_to_invoke",
+                                             "args" : dict_of_kwargs }
+
+    :returns: None
+    """
+    return _get_impl().cast(cfg.CONF, context, topic, msg)
+
+
+def fanout_cast(context, topic, msg):
+    """Broadcast a remote method invocation with no return.
+
+    This method will get invoked on all consumers that were set up with this
+    topic name and fanout=True.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param topic: The topic to send the rpc message to.  This correlates to the
+                  topic argument of
+                  nova.rpc.common.Connection.create_consumer() and only applies
+                  when the consumer was created with fanout=True.
+    :param msg: This is a dict in the form { "method" : "method_to_invoke",
+                                             "args" : dict_of_kwargs }
+
+    :returns: None
+    """
+    return _get_impl().fanout_cast(cfg.CONF, context, topic, msg)
+
+
+def multicall(context, topic, msg, timeout=None):
+    """Invoke a remote method and get back an iterator.
+
+    In this case, the remote method will be returning multiple values in
+    separate messages, so the return values can be processed as the come in via
+    an iterator.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param topic: The topic to send the rpc message to.  This correlates to the
+                  topic argument of
+                  nova.rpc.common.Connection.create_consumer() and only applies
+                  when the consumer was created with fanout=False.
+    :param msg: This is a dict in the form { "method" : "method_to_invoke",
+                                             "args" : dict_of_kwargs }
+    :param timeout: int, number of seconds to use for a response timeout.
+                    If set, this overrides the rpc_response_timeout option.
+
+    :returns: An iterator.  The iterator will yield a tuple (N, X) where N is
+              an index that starts at 0 and increases by one for each value
+              returned and X is the Nth value that was returned by the remote
+              method.
+
+    :raises: nova.rpc.common.Timeout if a complete response is not received
+             before the timeout is reached.
+    """
+    return _get_impl().multicall(cfg.CONF, context, topic, msg, timeout)
+
+
+def notify(context, topic, msg):
+    """Send notification event.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param topic: The topic to send the notification to.
+    :param msg: This is a dict of content of event.
+
+    :returns: None
+    """
+    return _get_impl().notify(cfg.CONF, context, topic, msg)
+
+
+def cleanup():
+    """Clean up resoruces in use by implementation.
+
+    Clean up any resources that have been allocated by the RPC implementation.
+    This is typically open connections to a messaging service.  This function
+    would get called before an application using this API exits to allow
+    connections to get torn down cleanly.
+
+    :returns: None
+    """
+    return _get_impl().cleanup()
+
+
+def cast_to_server(context, server_params, topic, msg):
+    """Invoke a remote method that does not return anything.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param server_params: Connection information
+    :param topic: The topic to send the notification to.
+    :param msg: This is a dict in the form { "method" : "method_to_invoke",
+                                             "args" : dict_of_kwargs }
+
+    :returns: None
+    """
+    return _get_impl().cast_to_server(cfg.CONF, context, server_params, topic,
+                                      msg)
+
+
+def fanout_cast_to_server(context, server_params, topic, msg):
+    """Broadcast to a remote method invocation with no return.
+
+    :param context: Information that identifies the user that has made this
+                    request.
+    :param server_params: Connection information
+    :param topic: The topic to send the notification to.
+    :param msg: This is a dict in the form { "method" : "method_to_invoke",
+                                             "args" : dict_of_kwargs }
+
+    :returns: None
+    """
+    return _get_impl().fanout_cast_to_server(cfg.CONF, context, server_params,
+                                             topic, msg)
+
+
+def queue_get_for(context, topic, host):
+    """Get a queue name for a given topic + host.
+
+    This function only works if this naming convention is followed on the
+    consumer side, as well.  For example, in nova, every instance of the
+    nova-foo service calls create_consumer() for two topics:
+
+        foo
+        foo.<host>
+
+    Messages sent to the 'foo' topic are distributed to exactly one instance of
+    the nova-foo service.  The services are chosen in a round-robin fashion.
+    Messages sent to the 'foo.<host>' topic are sent to the nova-foo service on
+    <host>.
+    """
+    return '%s.%s' % (topic, host)
+
+
+_RPCIMPL = None
+
+
+def _get_impl():
+    """Delay import of rpc_backend until configuration is loaded."""
+    global _RPCIMPL
+    if _RPCIMPL is None:
+        _RPCIMPL = importutils.import_module(cfg.CONF.rpc_backend)
+    return _RPCIMPL
diff --git a/openstack/common/rpc/amqp.py b/openstack/common/rpc/amqp.py
new file mode 100644
index 0000000..a79a3aa
--- /dev/null
+++ b/openstack/common/rpc/amqp.py
@@ -0,0 +1,416 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2010 United States Government as represented by the
+# Administrator of the National Aeronautics and Space Administration.
+# All Rights Reserved.
+# Copyright 2011 - 2012, Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+Shared code between AMQP based nova.rpc implementations.
+
+The code in this module is shared between the rpc implemenations based on AMQP.
+Specifically, this includes impl_kombu and impl_qpid.  impl_carrot also uses
+AMQP, but is deprecated and predates this code.
+"""
+
+import inspect
+import logging
+import sys
+import uuid
+
+from eventlet import greenpool
+from eventlet import pools
+from eventlet import semaphore
+
+from openstack.common import excutils
+from openstack.common import local
+import openstack.common.rpc.common as rpc_common
+
+
+LOG = logging.getLogger(__name__)
+
+
+class Pool(pools.Pool):
+    """Class that implements a Pool of Connections."""
+    def __init__(self, conf, connection_cls, *args, **kwargs):
+        self.connection_cls = connection_cls
+        self.conf = conf
+        kwargs.setdefault("max_size", self.conf.rpc_conn_pool_size)
+        kwargs.setdefault("order_as_stack", True)
+        super(Pool, self).__init__(*args, **kwargs)
+
+    # TODO(comstud): Timeout connections not used in a while
+    def create(self):
+        LOG.debug('Pool creating new connection')
+        return self.connection_cls(self.conf)
+
+    def empty(self):
+        while self.free_items:
+            self.get().close()
+
+
+_pool_create_sem = semaphore.Semaphore()
+
+
+def get_connection_pool(conf, connection_cls):
+    with _pool_create_sem:
+        # Make sure only one thread tries to create the connection pool.
+        if not connection_cls.pool:
+            connection_cls.pool = Pool(conf, connection_cls)
+    return connection_cls.pool
+
+
+class ConnectionContext(rpc_common.Connection):
+    """The class that is actually returned to the caller of
+    create_connection().  This is essentially a wrapper around
+    Connection that supports 'with'.  It can also return a new
+    Connection, or one from a pool.  The function will also catch
+    when an instance of this class is to be deleted.  With that
+    we can return Connections to the pool on exceptions and so
+    forth without making the caller be responsible for catching
+    them.  If possible the function makes sure to return a
+    connection to the pool.
+    """
+
+    def __init__(self, conf, connection_pool, pooled=True, server_params=None):
+        """Create a new connection, or get one from the pool"""
+        self.connection = None
+        self.conf = conf
+        self.connection_pool = connection_pool
+        if pooled:
+            self.connection = connection_pool.get()
+        else:
+            self.connection = connection_pool.connection_cls(conf,
+                    server_params=server_params)
+        self.pooled = pooled
+
+    def __enter__(self):
+        """When with ConnectionContext() is used, return self"""
+        return self
+
+    def _done(self):
+        """If the connection came from a pool, clean it up and put it back.
+        If it did not come from a pool, close it.
+        """
+        if self.connection:
+            if self.pooled:
+                # Reset the connection so it's ready for the next caller
+                # to grab from the pool
+                self.connection.reset()
+                self.connection_pool.put(self.connection)
+            else:
+                try:
+                    self.connection.close()
+                except Exception:
+                    pass
+            self.connection = None
+
+    def __exit__(self, exc_type, exc_value, tb):
+        """End of 'with' statement.  We're done here."""
+        self._done()
+
+    def __del__(self):
+        """Caller is done with this connection.  Make sure we cleaned up."""
+        self._done()
+
+    def close(self):
+        """Caller is done with this connection."""
+        self._done()
+
+    def create_consumer(self, topic, proxy, fanout=False):
+        self.connection.create_consumer(topic, proxy, fanout)
+
+    def create_worker(self, topic, proxy, pool_name):
+        self.connection.create_worker(topic, proxy, pool_name)
+
+    def consume_in_thread(self):
+        self.connection.consume_in_thread()
+
+    def __getattr__(self, key):
+        """Proxy all other calls to the Connection instance"""
+        if self.connection:
+            return getattr(self.connection, key)
+        else:
+            raise rpc_common.InvalidRPCConnectionReuse()
+
+
+def msg_reply(conf, msg_id, connection_pool, reply=None, failure=None,
+              ending=False):
+    """Sends a reply or an error on the channel signified by msg_id.
+
+    Failure should be a sys.exc_info() tuple.
+
+    """
+    with ConnectionContext(conf, connection_pool) as conn:
+        if failure:
+            failure = rpc_common.serialize_remote_exception(failure)
+
+        try:
+            msg = {'result': reply, 'failure': failure}
+        except TypeError:
+            msg = {'result': dict((k, repr(v))
+                            for k, v in reply.__dict__.iteritems()),
+                    'failure': failure}
+        if ending:
+            msg['ending'] = True
+        conn.direct_send(msg_id, msg)
+
+
+class RpcContext(rpc_common.CommonRpcContext):
+    """Context that supports replying to a rpc.call"""
+    def __init__(self, **kwargs):
+        self.msg_id = kwargs.pop('msg_id', None)
+        self.conf = kwargs.pop('conf')
+        super(RpcContext, self).__init__(**kwargs)
+
+    def deepcopy(self):
+        values = self.to_dict()
+        values['conf'] = self.conf
+        values['msg_id'] = self.msg_id
+        return self.__class__(**values)
+
+    def reply(self, reply=None, failure=None, ending=False,
+              connection_pool=None):
+        if self.msg_id:
+            msg_reply(self.conf, self.msg_id, connection_pool, reply, failure,
+                      ending)
+            if ending:
+                self.msg_id = None
+
+
+def unpack_context(conf, msg):
+    """Unpack context from msg."""
+    context_dict = {}
+    for key in list(msg.keys()):
+        # NOTE(vish): Some versions of python don't like unicode keys
+        #             in kwargs.
+        key = str(key)
+        if key.startswith('_context_'):
+            value = msg.pop(key)
+            context_dict[key[9:]] = value
+    context_dict['msg_id'] = msg.pop('_msg_id', None)
+    context_dict['conf'] = conf
+    ctx = RpcContext.from_dict(context_dict)
+    rpc_common._safe_log(LOG.debug, _('unpacked context: %s'), ctx.to_dict())
+    return ctx
+
+
+def pack_context(msg, context):
+    """Pack context into msg.
+
+    Values for message keys need to be less than 255 chars, so we pull
+    context out into a bunch of separate keys. If we want to support
+    more arguments in rabbit messages, we may want to do the same
+    for args at some point.
+
+    """
+    context_d = dict([('_context_%s' % key, value)
+                      for (key, value) in context.to_dict().iteritems()])
+    msg.update(context_d)
+
+
+class ProxyCallback(object):
+    """Calls methods on a proxy object based on method and args."""
+
+    def __init__(self, conf, proxy, connection_pool):
+        self.proxy = proxy
+        self.pool = greenpool.GreenPool(conf.rpc_thread_pool_size)
+        self.connection_pool = connection_pool
+        self.conf = conf
+
+    def __call__(self, message_data):
+        """Consumer callback to call a method on a proxy object.
+
+        Parses the message for validity and fires off a thread to call the
+        proxy object method.
+
+        Message data should be a dictionary with two keys:
+            method: string representing the method to call
+            args: dictionary of arg: value
+
+        Example: {'method': 'echo', 'args': {'value': 42}}
+
+        """
+        # It is important to clear the context here, because at this point
+        # the previous context is stored in local.store.context
+        if hasattr(local.store, 'context'):
+            del local.store.context
+        rpc_common._safe_log(LOG.debug, _('received %s'), message_data)
+        ctxt = unpack_context(self.conf, message_data)
+        method = message_data.get('method')
+        args = message_data.get('args', {})
+        version = message_data.get('version', None)
+        if not method:
+            LOG.warn(_('no method for message: %s') % message_data)
+            ctxt.reply(_('No method for message: %s') % message_data,
+                       connection_pool=self.connection_pool)
+            return
+        self.pool.spawn_n(self._process_data, ctxt, version, method, args)
+
+    def _process_data(self, ctxt, version, method, args):
+        """Process a message in a new thread.
+
+        If the proxy object we have has a dispatch method
+        (see rpc.dispatcher.RpcDispatcher), pass it the version,
+        method, and args and let it dispatch as appropriate.  If not, use
+        the old behavior of magically calling the specified method on the
+        proxy we have here.
+        """
+        ctxt.update_store()
+        try:
+            rval = self.proxy.dispatch(ctxt, version, method, **args)
+            # Check if the result was a generator
+            if inspect.isgenerator(rval):
+                for x in rval:
+                    ctxt.reply(x, None, connection_pool=self.connection_pool)
+            else:
+                ctxt.reply(rval, None, connection_pool=self.connection_pool)
+            # This final None tells multicall that it is done.
+            ctxt.reply(ending=True, connection_pool=self.connection_pool)
+        except Exception as e:
+            LOG.exception('Exception during message handling')
+            ctxt.reply(None, sys.exc_info(),
+                       connection_pool=self.connection_pool)
+
+
+class MulticallWaiter(object):
+    def __init__(self, conf, connection, timeout):
+        self._connection = connection
+        self._iterator = connection.iterconsume(
+                                timeout=timeout or conf.rpc_response_timeout)
+        self._result = None
+        self._done = False
+        self._got_ending = False
+        self._conf = conf
+
+    def done(self):
+        if self._done:
+            return
+        self._done = True
+        self._iterator.close()
+        self._iterator = None
+        self._connection.close()
+
+    def __call__(self, data):
+        """The consume() callback will call this.  Store the result."""
+        if data['failure']:
+            failure = data['failure']
+            self._result = rpc_common.deserialize_remote_exception(self._conf,
+                    failure)
+
+        elif data.get('ending', False):
+            self._got_ending = True
+        else:
+            self._result = data['result']
+
+    def __iter__(self):
+        """Return a result until we get a 'None' response from consumer"""
+        if self._done:
+            raise StopIteration
+        while True:
+            try:
+                self._iterator.next()
+            except Exception:
+                with excutils.save_and_reraise_exception():
+                    self.done()
+            if self._got_ending:
+                self.done()
+                raise StopIteration
+            result = self._result
+            if isinstance(result, Exception):
+                self.done()
+                raise result
+            yield result
+
+
+def create_connection(conf, new, connection_pool):
+    """Create a connection"""
+    return ConnectionContext(conf, connection_pool, pooled=not new)
+
+
+def multicall(conf, context, topic, msg, timeout, connection_pool):
+    """Make a call that returns multiple times."""
+    # Can't use 'with' for multicall, as it returns an iterator
+    # that will continue to use the connection.  When it's done,
+    # connection.close() will get called which will put it back into
+    # the pool
+    LOG.debug(_('Making asynchronous call on %s ...'), topic)
+    msg_id = uuid.uuid4().hex
+    msg.update({'_msg_id': msg_id})
+    LOG.debug(_('MSG_ID is %s') % (msg_id))
+    pack_context(msg, context)
+
+    conn = ConnectionContext(conf, connection_pool)
+    wait_msg = MulticallWaiter(conf, conn, timeout)
+    conn.declare_direct_consumer(msg_id, wait_msg)
+    conn.topic_send(topic, msg)
+    return wait_msg
+
+
+def call(conf, context, topic, msg, timeout, connection_pool):
+    """Sends a message on a topic and wait for a response."""
+    rv = multicall(conf, context, topic, msg, timeout, connection_pool)
+    # NOTE(vish): return the last result from the multicall
+    rv = list(rv)
+    if not rv:
+        return
+    return rv[-1]
+
+
+def cast(conf, context, topic, msg, connection_pool):
+    """Sends a message on a topic without waiting for a response."""
+    LOG.debug(_('Making asynchronous cast on %s...'), topic)
+    pack_context(msg, context)
+    with ConnectionContext(conf, connection_pool) as conn:
+        conn.topic_send(topic, msg)
+
+
+def fanout_cast(conf, context, topic, msg, connection_pool):
+    """Sends a message on a fanout exchange without waiting for a response."""
+    LOG.debug(_('Making asynchronous fanout cast...'))
+    pack_context(msg, context)
+    with ConnectionContext(conf, connection_pool) as conn:
+        conn.fanout_send(topic, msg)
+
+
+def cast_to_server(conf, context, server_params, topic, msg, connection_pool):
+    """Sends a message on a topic to a specific server."""
+    pack_context(msg, context)
+    with ConnectionContext(conf, connection_pool, pooled=False,
+            server_params=server_params) as conn:
+        conn.topic_send(topic, msg)
+
+
+def fanout_cast_to_server(conf, context, server_params, topic, msg,
+        connection_pool):
+    """Sends a message on a fanout exchange to a specific server."""
+    pack_context(msg, context)
+    with ConnectionContext(conf, connection_pool, pooled=False,
+            server_params=server_params) as conn:
+        conn.fanout_send(topic, msg)
+
+
+def notify(conf, context, topic, msg, connection_pool):
+    """Sends a notification event on a topic."""
+    event_type = msg.get('event_type')
+    LOG.debug(_('Sending %(event_type)s on %(topic)s'), locals())
+    pack_context(msg, context)
+    with ConnectionContext(conf, connection_pool) as conn:
+        conn.notify_send(topic, msg)
+
+
+def cleanup(connection_pool):
+    if connection_pool:
+        connection_pool.empty()
diff --git a/openstack/common/rpc/common.py b/openstack/common/rpc/common.py
new file mode 100644
index 0000000..6acd72c
--- /dev/null
+++ b/openstack/common/rpc/common.py
@@ -0,0 +1,316 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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 openstack.common import cfg
+from openstack.common import importutils
+from openstack.common import jsonutils
+from openstack.common import local
+
+
+LOG = logging.getLogger(__name__)
+
+
+class RPCException(Exception):
+    message = _("An unknown RPC related exception occurred.")
+
+    def __init__(self, message=None, **kwargs):
+        self.kwargs = kwargs
+
+        if not message:
+            try:
+                message = self.message % kwargs
+
+            except Exception as e:
+                # 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 kwargs.iteritems():
+                    LOG.error("%s: %s" % (name, value))
+                # at least get the core message out if something happened
+                message = self.message
+
+        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.
+
+    """
+    message = _("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.
+    """
+    message = _("Timeout while waiting on RPC response.")
+
+
+class InvalidRPCConnectionReuse(RPCException):
+    message = _("Invalid reuse of an RPC connection.")
+
+
+class UnsupportedRpcVersion(RPCException):
+    message = _("Specified RPC version, %(version)s, not supported by "
+                "this endpoint.")
+
+
+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, conf, 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 conf:  An openstack.common.cfg configuration 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, conf, 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 conf:  An openstack.common.cfg configuration object.
+        :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 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 = {
+                'set_admin_password': ('new_pass',),
+                'run_instance': ('admin_password',),
+               }
+
+    has_method = 'method' in msg_data and msg_data['method'] in SANITIZE
+    has_context_token = '_context_auth_token' in msg_data
+    has_token = 'auth_token' in msg_data
+
+    if not any([has_method, has_context_token, has_token]):
+        return log_func(msg, msg_data)
+
+    msg_data = copy.deepcopy(msg_data)
+
+    if has_method:
+        method = msg_data['method']
+        if method in SANITIZE:
+            args_to_sanitize = SANITIZE[method]
+            for arg in args_to_sanitize:
+                try:
+                    msg_data['args'][arg] = "<SANITIZED>"
+                except KeyError:
+                    pass
+
+    if has_context_token:
+        msg_data['_context_auth_token'] = '<SANITIZED>'
+
+    if has_token:
+        msg_data['auth_token'] = '<SANITIZED>'
+
+    return log_func(msg, msg_data)
+
+
+def serialize_remote_exception(failure_info):
+    """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]
+    LOG.error(_("Returning exception %s to caller"), unicode(failure))
+    LOG.error(tb)
+
+    kwargs = {}
+    if hasattr(failure, 'kwargs'):
+        kwargs = failure.kwargs
+
+    data = {
+        'class': str(failure.__class__.__name__),
+        'module': str(failure.__class__.__module__),
+        'message': unicode(failure),
+        'tb': tb,
+        'args': failure.args,
+        'kwargs': kwargs
+    }
+
+    json_data = jsonutils.dumps(data)
+
+    return json_data
+
+
+def deserialize_remote_exception(conf, data):
+    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 not module in conf.allowed_rpc_exception_modules:
+        return 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('kwargs', {}))
+    except (AttributeError, TypeError, ImportError):
+        return RemoteError(name, failure.get('message'), trace)
+
+    ex_type = type(failure)
+    str_override = lambda self: message
+    new_ex_type = type(ex_type.__name__ + "_Remote", (ex_type,),
+                       {'__str__': str_override, '__unicode__': str_override})
+    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 as e:
+        # 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
+
+    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
diff --git a/openstack/common/rpc/dispatcher.py b/openstack/common/rpc/dispatcher.py
new file mode 100644
index 0000000..7319eb2
--- /dev/null
+++ b/openstack/common/rpc/dispatcher.py
@@ -0,0 +1,105 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+Code for rpc message dispatching.
+
+Messages that come in have a version number associated with them.  RPC API
+version numbers are in the form:
+
+    Major.Minor
+
+For a given message with version X.Y, the receiver must be marked as able to
+handle messages of version A.B, where:
+
+    A = X
+
+    B >= Y
+
+The Major version number would be incremented for an almost completely new API.
+The Minor version number would be incremented for backwards compatible changes
+to an existing API.  A backwards compatible change could be something like
+adding a new method, adding an argument to an existing method (but not
+requiring it), or changing the type for an existing argument (but still
+handling the old type as well).
+
+The conversion over to a versioned API must be done on both the client side and
+server side of the API at the same time.  However, as the code stands today,
+there can be both versioned and unversioned APIs implemented in the same code
+base.
+"""
+
+from openstack.common.rpc import common as rpc_common
+
+
+class RpcDispatcher(object):
+    """Dispatch rpc messages according to the requested API version.
+
+    This class can be used as the top level 'manager' for a service.  It
+    contains a list of underlying managers that have an API_VERSION attribute.
+    """
+
+    def __init__(self, callbacks):
+        """Initialize the rpc dispatcher.
+
+        :param callbacks: List of proxy objects that are an instance
+                          of a class with rpc methods exposed.  Each proxy
+                          object should have an RPC_API_VERSION attribute.
+        """
+        self.callbacks = callbacks
+        super(RpcDispatcher, self).__init__()
+
+    @staticmethod
+    def _is_compatible(mversion, version):
+        """Determine whether versions are compatible.
+
+        :param mversion: The API version implemented by a callback.
+        :param version: The API version requested by an incoming message.
+        """
+        version_parts = version.split('.')
+        mversion_parts = mversion.split('.')
+        if int(version_parts[0]) != int(mversion_parts[0]):  # Major
+            return False
+        if int(version_parts[1]) > int(mversion_parts[1]):  # Minor
+            return False
+        return True
+
+    def dispatch(self, ctxt, version, method, **kwargs):
+        """Dispatch a message based on a requested version.
+
+        :param ctxt: The request context
+        :param version: The requested API version from the incoming message
+        :param method: The method requested to be called by the incoming
+                       message.
+        :param kwargs: A dict of keyword arguments to be passed to the method.
+
+        :returns: Whatever is returned by the underlying method that gets
+                  called.
+        """
+        if not version:
+            version = '1.0'
+
+        for proxyobj in self.callbacks:
+            if hasattr(proxyobj, 'RPC_API_VERSION'):
+                rpc_api_version = proxyobj.RPC_API_VERSION
+            else:
+                rpc_api_version = '1.0'
+            if not hasattr(proxyobj, method):
+                continue
+            if self._is_compatible(rpc_api_version, version):
+                return getattr(proxyobj, method)(ctxt, **kwargs)
+
+        raise rpc_common.UnsupportedRpcVersion(version=version)
diff --git a/openstack/common/rpc/impl_fake.py b/openstack/common/rpc/impl_fake.py
new file mode 100644
index 0000000..fba20c9
--- /dev/null
+++ b/openstack/common/rpc/impl_fake.py
@@ -0,0 +1,184 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+#    Copyright 2011 OpenStack LLC
+#
+#    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.
+"""Fake RPC implementation which calls proxy methods directly with no
+queues.  Casts will block, but this is very useful for tests.
+"""
+
+import inspect
+import json
+import time
+
+import eventlet
+
+from openstack.common.rpc import common as rpc_common
+
+CONSUMERS = {}
+
+
+class RpcContext(rpc_common.CommonRpcContext):
+    def __init__(self, **kwargs):
+        super(RpcContext, self).__init__(**kwargs)
+        self._response = []
+        self._done = False
+
+    def deepcopy(self):
+        values = self.to_dict()
+        new_inst = self.__class__(**values)
+        new_inst._response = self._response
+        new_inst._done = self._done
+        return new_inst
+
+    def reply(self, reply=None, failure=None, ending=False):
+        if ending:
+            self._done = True
+        if not self._done:
+            self._response.append((reply, failure))
+
+
+class Consumer(object):
+    def __init__(self, topic, proxy):
+        self.topic = topic
+        self.proxy = proxy
+
+    def call(self, context, version, method, args, timeout):
+        done = eventlet.event.Event()
+
+        def _inner():
+            ctxt = RpcContext.from_dict(context.to_dict())
+            try:
+                rval = self.proxy.dispatch(context, version, method, **args)
+                res = []
+                # Caller might have called ctxt.reply() manually
+                for (reply, failure) in ctxt._response:
+                    if failure:
+                        raise failure[0], failure[1], failure[2]
+                    res.append(reply)
+                # if ending not 'sent'...we might have more data to
+                # return from the function itself
+                if not ctxt._done:
+                    if inspect.isgenerator(rval):
+                        for val in rval:
+                            res.append(val)
+                    else:
+                        res.append(rval)
+                done.send(res)
+            except Exception as e:
+                done.send_exception(e)
+
+        thread = eventlet.greenthread.spawn(_inner)
+
+        if timeout:
+            start_time = time.time()
+            while not done.ready():
+                eventlet.greenthread.sleep(1)
+                cur_time = time.time()
+                if (cur_time - start_time) > timeout:
+                    thread.kill()
+                    raise rpc_common.Timeout()
+
+        return done.wait()
+
+
+class Connection(object):
+    """Connection object."""
+
+    def __init__(self):
+        self.consumers = []
+
+    def create_consumer(self, topic, proxy, fanout=False):
+        consumer = Consumer(topic, proxy)
+        self.consumers.append(consumer)
+        if topic not in CONSUMERS:
+            CONSUMERS[topic] = []
+        CONSUMERS[topic].append(consumer)
+
+    def close(self):
+        for consumer in self.consumers:
+            CONSUMERS[consumer.topic].remove(consumer)
+        self.consumers = []
+
+    def consume_in_thread(self):
+        pass
+
+
+def create_connection(conf, new=True):
+    """Create a connection"""
+    return Connection()
+
+
+def check_serialize(msg):
+    """Make sure a message intended for rpc can be serialized."""
+    json.dumps(msg)
+
+
+def multicall(conf, context, topic, msg, timeout=None):
+    """Make a call that returns multiple times."""
+
+    check_serialize(msg)
+
+    method = msg.get('method')
+    if not method:
+        return
+    args = msg.get('args', {})
+    version = msg.get('version', None)
+
+    try:
+        consumer = CONSUMERS[topic][0]
+    except (KeyError, IndexError):
+        return iter([None])
+    else:
+        return consumer.call(context, version, method, args, timeout)
+
+
+def call(conf, context, topic, msg, timeout=None):
+    """Sends a message on a topic and wait for a response."""
+    rv = multicall(conf, context, topic, msg, timeout)
+    # NOTE(vish): return the last result from the multicall
+    rv = list(rv)
+    if not rv:
+        return
+    return rv[-1]
+
+
+def cast(conf, context, topic, msg):
+    try:
+        call(conf, context, topic, msg)
+    except Exception:
+        pass
+
+
+def notify(conf, context, topic, msg):
+    check_serialize(msg)
+
+
+def cleanup():
+    pass
+
+
+def fanout_cast(conf, context, topic, msg):
+    """Cast to all consumers of a topic"""
+    check_serialize(msg)
+    method = msg.get('method')
+    if not method:
+        return
+    args = msg.get('args', {})
+    version = msg.get('version', None)
+
+    for consumer in CONSUMERS.get(topic, []):
+        try:
+            consumer.call(context, version, method, args, None)
+        except Exception:
+            pass
diff --git a/openstack/common/rpc/impl_kombu.py b/openstack/common/rpc/impl_kombu.py
new file mode 100644
index 0000000..c32497a
--- /dev/null
+++ b/openstack/common/rpc/impl_kombu.py
@@ -0,0 +1,758 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+#    Copyright 2011 OpenStack LLC
+#
+#    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 socket
+import ssl
+import sys
+import time
+import uuid
+
+import eventlet
+import greenlet
+import kombu
+import kombu.connection
+import kombu.entity
+import kombu.messaging
+
+from openstack.common import cfg
+from openstack.common.rpc import amqp as rpc_amqp
+from openstack.common.rpc import common as rpc_common
+
+kombu_opts = [
+    cfg.StrOpt('kombu_ssl_version',
+               default='',
+               help='SSL version to use (valid only if SSL enabled)'),
+    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.StrOpt('rabbit_host',
+               default='localhost',
+               help='the RabbitMQ host'),
+    cfg.IntOpt('rabbit_port',
+               default=5672,
+               help='the RabbitMQ port'),
+    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'),
+    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 retries with trying to connect to RabbitMQ '
+                    '(the default of 0 implies an infinite retry count)'),
+    cfg.BoolOpt('rabbit_durable_queues',
+                default=False,
+                help='use durable queues in RabbitMQ'),
+
+    ]
+
+cfg.CONF.register_opts(kombu_opts)
+
+LOG = rpc_common.LOG
+
+
+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 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.
+
+        Messages will automatically be acked if the callback doesn't
+        raise an exception
+        """
+
+        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)
+            try:
+                callback(message.payload)
+                message.ack()
+            except Exception:
+                LOG.exception(_("Failed to process message... skipping it."))
+
+        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, 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,
+                'auto_delete': True,
+                'exclusive': True}
+        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,
+                 **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.rabbit_durable_queues,
+                'auto_delete': False,
+                'exclusive': False}
+        options.update(kwargs)
+        exchange = kombu.entity.Exchange(
+                name=conf.control_exchange,
+                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,
+                'auto_delete': True,
+                'exclusive': True}
+        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):
+        """Send a message"""
+        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': 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.rabbit_durable_queues,
+                'auto_delete': False,
+                'exclusive': False}
+        options.update(kwargs)
+        super(TopicPublisher, self).__init__(channel,
+                conf.control_exchange,
+                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': True}
+        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.rabbit_durable_queues)
+        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.declare()
+
+
+class Connection(object):
+    """Connection object."""
+
+    pool = None
+
+    def __init__(self, conf, server_params=None):
+        self.consumers = []
+        self.consumer_thread = None
+        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'}
+
+        params = {}
+        for sp_key, value in server_params.iteritems():
+            p_key = server_params_to_kombu_params.get(sp_key, sp_key)
+            params[p_key] = value
+
+        params.setdefault('hostname', self.conf.rabbit_host)
+        params.setdefault('port', self.conf.rabbit_port)
+        params.setdefault('userid', self.conf.rabbit_userid)
+        params.setdefault('password', self.conf.rabbit_password)
+        params.setdefault('virtual_host', self.conf.rabbit_virtual_host)
+
+        self.params = params
+
+        if self.conf.fake_rabbit:
+            self.params['transport'] = 'memory'
+            self.memory_transport = True
+        else:
+            self.memory_transport = False
+
+        if self.conf.rabbit_use_ssl:
+            self.params['ssl'] = self._fetch_ssl_params()
+
+        self.connection = None
+        self.reconnect()
+
+    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.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
+
+        if not ssl_params:
+            # Just have the default behavior
+            return True
+        else:
+            # Return the extended behavior
+            return ssl_params
+
+    def _connect(self):
+        """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") % self.params)
+            try:
+                self.connection.close()
+            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(
+                **self.params)
+        self.connection_errors = self.connection.connection_errors
+        if self.memory_transport:
+            # Kludge to speed up tests.
+            self.connection.transport.polling_interval = 0.0
+        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'),
+                 self.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:
+            attempt += 1
+            try:
+                self._connect()
+                return
+            except (self.connection_errors, IOError), e:
+                pass
+            except Exception, 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(self.params)
+
+            if self.max_retries and attempt == self.max_retries:
+                LOG.exception(_('Unable to connect to AMQP server on '
+                        '%(hostname)s:%(port)d after %(max_retries)d '
+                        'tries: %(err_str)s') % log_info)
+                # NOTE(comstud): Copied from original code.  There's
+                # really no better recourse because if this was a queue we
+                # need to consume on, we have no way to consume anymore.
+                sys.exit(1)
+
+            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.exception(_('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, socket.timeout, IOError), e:
+                pass
+            except Exception, 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.cancel_consumer_thread()
+        self.connection.release()
+        self.connection = None
+
+    def reset(self):
+        """Reset a connection so it can be used again"""
+        self.cancel_consumer_thread()
+        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,
+                    self.consumer_num.next())
+            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"""
+
+        info = {'do_consume': True}
+
+        def _error_callback(exc):
+            if isinstance(exc, socket.timeout):
+                LOG.exception(_('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))
+                info['do_consume'] = True
+
+        def _consume():
+            if info['do_consume']:
+                queues_head = self.consumers[:-1]
+                queues_tail = self.consumers[-1]
+                for queue in queues_head:
+                    queue.consume(nowait=True)
+                queues_tail.consume(nowait=False)
+                info['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 cancel_consumer_thread(self):
+        """Cancel a consumer thread"""
+        if self.consumer_thread is not None:
+            self.consumer_thread.kill()
+            try:
+                self.consumer_thread.wait()
+            except greenlet.GreenletExit:
+                pass
+            self.consumer_thread = None
+
+    def publisher_send(self, cls, topic, msg, **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)
+
+        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):
+        """Create a 'topic' consumer."""
+        self.declare_consumer(functools.partial(TopicConsumer,
+                                                name=queue_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):
+        """Send a 'topic' message"""
+        self.publisher_send(TopicPublisher, topic, msg)
+
+    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, **kwargs)
+
+    def consume(self, limit=None):
+        """Consume from all queues/consumers"""
+        it = self.iterconsume(limit=limit)
+        while True:
+            try:
+                it.next()
+            except StopIteration:
+                return
+
+    def consume_in_thread(self):
+        """Consumer from all queues/consumers in a greenthread"""
+        def _consumer_thread():
+            try:
+                self.consume()
+            except greenlet.GreenletExit:
+                return
+        if self.consumer_thread is None:
+            self.consumer_thread = eventlet.spawn(_consumer_thread)
+        return self.consumer_thread
+
+    def create_consumer(self, topic, proxy, fanout=False):
+        """Create a consumer that calls a method in a proxy object"""
+        proxy_cb = rpc_amqp.ProxyCallback(self.conf, proxy,
+                rpc_amqp.get_connection_pool(self.conf, Connection))
+
+        if fanout:
+            self.declare_fanout_consumer(topic, proxy_cb)
+        else:
+            self.declare_topic_consumer(topic, proxy_cb)
+
+    def create_worker(self, topic, proxy, pool_name):
+        """Create a worker that calls a method in a proxy object"""
+        proxy_cb = rpc_amqp.ProxyCallback(self.conf, proxy,
+                rpc_amqp.get_connection_pool(self.conf, Connection))
+        self.declare_topic_consumer(topic, proxy_cb, pool_name)
+
+
+def create_connection(conf, new=True):
+    """Create a connection"""
+    return rpc_amqp.create_connection(conf, new,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def multicall(conf, context, topic, msg, timeout=None):
+    """Make a call that returns multiple times."""
+    return rpc_amqp.multicall(conf, context, topic, msg, timeout,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def call(conf, context, topic, msg, timeout=None):
+    """Sends a message on a topic and wait for a response."""
+    return rpc_amqp.call(conf, context, topic, msg, timeout,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def cast(conf, context, topic, msg):
+    """Sends a message on a topic without waiting for a response."""
+    return rpc_amqp.cast(conf, context, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def fanout_cast(conf, context, topic, msg):
+    """Sends a message on a fanout exchange without waiting for a response."""
+    return rpc_amqp.fanout_cast(conf, context, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def cast_to_server(conf, context, server_params, topic, msg):
+    """Sends a message on a topic to a specific server."""
+    return rpc_amqp.cast_to_server(conf, context, server_params, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def fanout_cast_to_server(conf, context, server_params, topic, msg):
+    """Sends a message on a fanout exchange to a specific server."""
+    return rpc_amqp.cast_to_server(conf, context, server_params, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def notify(conf, context, topic, msg):
+    """Sends a notification event on a topic."""
+    return rpc_amqp.notify(conf, context, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def cleanup():
+    return rpc_amqp.cleanup(Connection.pool)
diff --git a/openstack/common/rpc/impl_qpid.py b/openstack/common/rpc/impl_qpid.py
new file mode 100644
index 0000000..3c46309
--- /dev/null
+++ b/openstack/common/rpc/impl_qpid.py
@@ -0,0 +1,580 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+#    Copyright 2011 OpenStack LLC
+#    Copyright 2011 - 2012, Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import functools
+import itertools
+import json
+import logging
+import time
+import uuid
+
+import eventlet
+import greenlet
+import qpid.messaging
+import qpid.messaging.exceptions
+
+from openstack.common import cfg
+from openstack.common.gettextutils import _
+from openstack.common.rpc import amqp as rpc_amqp
+from openstack.common.rpc import common as rpc_common
+
+LOG = logging.getLogger(__name__)
+
+qpid_opts = [
+    cfg.StrOpt('qpid_hostname',
+               default='localhost',
+               help='Qpid broker hostname'),
+    cfg.StrOpt('qpid_port',
+               default='5672',
+               help='Qpid broker port'),
+    cfg.StrOpt('qpid_username',
+               default='',
+               help='Username for qpid connection'),
+    cfg.StrOpt('qpid_password',
+               default='',
+               help='Password for qpid connection'),
+    cfg.StrOpt('qpid_sasl_mechanisms',
+               default='',
+               help='Space separated list of SASL mechanisms to use for auth'),
+    cfg.BoolOpt('qpid_reconnect',
+                default=True,
+                help='Automatically reconnect'),
+    cfg.IntOpt('qpid_reconnect_timeout',
+               default=0,
+               help='Reconnection timeout in seconds'),
+    cfg.IntOpt('qpid_reconnect_limit',
+               default=0,
+               help='Max reconnections before giving up'),
+    cfg.IntOpt('qpid_reconnect_interval_min',
+               default=0,
+               help='Minimum seconds between reconnection attempts'),
+    cfg.IntOpt('qpid_reconnect_interval_max',
+               default=0,
+               help='Maximum seconds between reconnection attempts'),
+    cfg.IntOpt('qpid_reconnect_interval',
+               default=0,
+               help='Equivalent to setting max and min to the same value'),
+    cfg.IntOpt('qpid_heartbeat',
+               default=5,
+               help='Seconds between connection keepalive heartbeats'),
+    cfg.StrOpt('qpid_protocol',
+               default='tcp',
+               help="Transport to use, either 'tcp' or 'ssl'"),
+    cfg.BoolOpt('qpid_tcp_nodelay',
+                default=True,
+                help='Disable Nagle algorithm'),
+    ]
+
+cfg.CONF.register_opts(qpid_opts)
+
+
+class ConsumerBase(object):
+    """Consumer base class."""
+
+    def __init__(self, session, callback, node_name, node_opts,
+                 link_name, link_opts):
+        """Declare a queue on an amqp session.
+
+        'session' is the amqp session to use
+        'callback' is the callback to call when messages are received
+        'node_name' is the first part of the Qpid address string, before ';'
+        'node_opts' will be applied to the "x-declare" section of "node"
+                    in the address string.
+        'link_name' goes into the "name" field of the "link" in the address
+                    string
+        'link_opts' will be applied to the "x-declare" section of "link"
+                    in the address string.
+        """
+        self.callback = callback
+        self.receiver = None
+        self.session = None
+
+        addr_opts = {
+            "create": "always",
+            "node": {
+                "type": "topic",
+                "x-declare": {
+                    "durable": True,
+                    "auto-delete": True,
+                },
+            },
+            "link": {
+                "name": link_name,
+                "durable": True,
+                "x-declare": {
+                    "durable": False,
+                    "auto-delete": True,
+                    "exclusive": False,
+                },
+            },
+        }
+        addr_opts["node"]["x-declare"].update(node_opts)
+        addr_opts["link"]["x-declare"].update(link_opts)
+
+        self.address = "%s ; %s" % (node_name, json.dumps(addr_opts))
+
+        self.reconnect(session)
+
+    def reconnect(self, session):
+        """Re-declare the receiver after a qpid reconnect"""
+        self.session = session
+        self.receiver = session.receiver(self.address)
+        self.receiver.capacity = 1
+
+    def consume(self):
+        """Fetch the message and pass it to the callback object"""
+        message = self.receiver.fetch()
+        self.callback(message.content)
+
+    def get_receiver(self):
+        return self.receiver
+
+
+class DirectConsumer(ConsumerBase):
+    """Queue/consumer class for 'direct'"""
+
+    def __init__(self, conf, session, msg_id, callback):
+        """Init a 'direct' queue.
+
+        'session' is the amqp session to use
+        'msg_id' is the msg_id to listen on
+        'callback' is the callback to call when messages are received
+        """
+
+        super(DirectConsumer, self).__init__(session, callback,
+                        "%s/%s" % (msg_id, msg_id),
+                        {"type": "direct"},
+                        msg_id,
+                        {"exclusive": True})
+
+
+class TopicConsumer(ConsumerBase):
+    """Consumer class for 'topic'"""
+
+    def __init__(self, conf, session, topic, callback, name=None):
+        """Init a 'topic' queue.
+
+        :param session: the amqp session to use
+        :param topic: is the topic to listen on
+        :paramtype topic: str
+        :param callback: the callback to call when messages are received
+        :param name: optional queue name, defaults to topic
+        """
+
+        super(TopicConsumer, self).__init__(session, callback,
+                        "%s/%s" % (conf.control_exchange, topic), {},
+                        name or topic, {})
+
+
+class FanoutConsumer(ConsumerBase):
+    """Consumer class for 'fanout'"""
+
+    def __init__(self, conf, session, topic, callback):
+        """Init a 'fanout' queue.
+
+        'session' is the amqp session to use
+        'topic' is the topic to listen on
+        'callback' is the callback to call when messages are received
+        """
+
+        super(FanoutConsumer, self).__init__(session, callback,
+                        "%s_fanout" % topic,
+                        {"durable": False, "type": "fanout"},
+                        "%s_fanout_%s" % (topic, uuid.uuid4().hex),
+                        {"exclusive": True})
+
+
+class Publisher(object):
+    """Base Publisher class"""
+
+    def __init__(self, session, node_name, node_opts=None):
+        """Init the Publisher class with the exchange_name, routing_key,
+        and other options
+        """
+        self.sender = None
+        self.session = session
+
+        addr_opts = {
+            "create": "always",
+            "node": {
+                "type": "topic",
+                "x-declare": {
+                    "durable": False,
+                    # auto-delete isn't implemented for exchanges in qpid,
+                    # but put in here anyway
+                    "auto-delete": True,
+                },
+            },
+        }
+        if node_opts:
+            addr_opts["node"]["x-declare"].update(node_opts)
+
+        self.address = "%s ; %s" % (node_name, json.dumps(addr_opts))
+
+        self.reconnect(session)
+
+    def reconnect(self, session):
+        """Re-establish the Sender after a reconnection"""
+        self.sender = session.sender(self.address)
+
+    def send(self, msg):
+        """Send a message"""
+        self.sender.send(msg)
+
+
+class DirectPublisher(Publisher):
+    """Publisher class for 'direct'"""
+    def __init__(self, conf, session, msg_id):
+        """Init a 'direct' publisher."""
+        super(DirectPublisher, self).__init__(session, msg_id,
+                                              {"type": "Direct"})
+
+
+class TopicPublisher(Publisher):
+    """Publisher class for 'topic'"""
+    def __init__(self, conf, session, topic):
+        """init a 'topic' publisher.
+        """
+        super(TopicPublisher, self).__init__(session,
+                                "%s/%s" % (conf.control_exchange, topic))
+
+
+class FanoutPublisher(Publisher):
+    """Publisher class for 'fanout'"""
+    def __init__(self, conf, session, topic):
+        """init a 'fanout' publisher.
+        """
+        super(FanoutPublisher, self).__init__(session,
+                                "%s_fanout" % topic, {"type": "fanout"})
+
+
+class NotifyPublisher(Publisher):
+    """Publisher class for notifications"""
+    def __init__(self, conf, session, topic):
+        """init a 'topic' publisher.
+        """
+        super(NotifyPublisher, self).__init__(session,
+                                "%s/%s" % (conf.control_exchange, topic),
+                                {"durable": True})
+
+
+class Connection(object):
+    """Connection object."""
+
+    pool = None
+
+    def __init__(self, conf, server_params=None):
+        self.session = None
+        self.consumers = {}
+        self.consumer_thread = None
+        self.conf = conf
+
+        if server_params is None:
+            server_params = {}
+
+        default_params = dict(hostname=self.conf.qpid_hostname,
+                port=self.conf.qpid_port,
+                username=self.conf.qpid_username,
+                password=self.conf.qpid_password)
+
+        params = server_params
+        for key in default_params.keys():
+            params.setdefault(key, default_params[key])
+
+        self.broker = params['hostname'] + ":" + str(params['port'])
+        # Create the connection - this does not open the connection
+        self.connection = qpid.messaging.Connection(self.broker)
+
+        # Check if flags are set and if so set them for the connection
+        # before we call open
+        self.connection.username = params['username']
+        self.connection.password = params['password']
+        self.connection.sasl_mechanisms = self.conf.qpid_sasl_mechanisms
+        self.connection.reconnect = self.conf.qpid_reconnect
+        if self.conf.qpid_reconnect_timeout:
+            self.connection.reconnect_timeout = (
+                    self.conf.qpid_reconnect_timeout)
+        if self.conf.qpid_reconnect_limit:
+            self.connection.reconnect_limit = self.conf.qpid_reconnect_limit
+        if self.conf.qpid_reconnect_interval_max:
+            self.connection.reconnect_interval_max = (
+                    self.conf.qpid_reconnect_interval_max)
+        if self.conf.qpid_reconnect_interval_min:
+            self.connection.reconnect_interval_min = (
+                    self.conf.qpid_reconnect_interval_min)
+        if self.conf.qpid_reconnect_interval:
+            self.connection.reconnect_interval = (
+                    self.conf.qpid_reconnect_interval)
+        self.connection.hearbeat = self.conf.qpid_heartbeat
+        self.connection.protocol = self.conf.qpid_protocol
+        self.connection.tcp_nodelay = self.conf.qpid_tcp_nodelay
+
+        # Open is part of reconnect -
+        # NOTE(WGH) not sure we need this with the reconnect flags
+        self.reconnect()
+
+    def _register_consumer(self, consumer):
+        self.consumers[str(consumer.get_receiver())] = consumer
+
+    def _lookup_consumer(self, receiver):
+        return self.consumers[str(receiver)]
+
+    def reconnect(self):
+        """Handles reconnecting and re-establishing sessions and queues"""
+        if self.connection.opened():
+            try:
+                self.connection.close()
+            except qpid.messaging.exceptions.ConnectionError:
+                pass
+
+        while True:
+            try:
+                self.connection.open()
+            except qpid.messaging.exceptions.ConnectionError, e:
+                LOG.error(_('Unable to connect to AMQP server: %s'), e)
+                time.sleep(self.conf.qpid_reconnect_interval or 1)
+            else:
+                break
+
+        LOG.info(_('Connected to AMQP server on %s'), self.broker)
+
+        self.session = self.connection.session()
+
+        for consumer in self.consumers.itervalues():
+            consumer.reconnect(self.session)
+
+        if self.consumers:
+            LOG.debug(_("Re-established AMQP queues"))
+
+    def ensure(self, error_callback, method, *args, **kwargs):
+        while True:
+            try:
+                return method(*args, **kwargs)
+            except (qpid.messaging.exceptions.Empty,
+                    qpid.messaging.exceptions.ConnectionError), e:
+                if error_callback:
+                    error_callback(e)
+                self.reconnect()
+
+    def close(self):
+        """Close/release this connection"""
+        self.cancel_consumer_thread()
+        self.connection.close()
+        self.connection = None
+
+    def reset(self):
+        """Reset a connection so it can be used again"""
+        self.cancel_consumer_thread()
+        self.session.close()
+        self.session = self.connection.session()
+        self.consumers = {}
+
+    def declare_consumer(self, consumer_cls, topic, callback):
+        """Create a Consumer using the class that was passed in and
+        add it to our list of consumers
+        """
+        def _connect_error(exc):
+            log_info = {'topic': topic, 'err_str': str(exc)}
+            LOG.error(_("Failed to declare consumer for topic '%(topic)s': "
+                "%(err_str)s") % log_info)
+
+        def _declare_consumer():
+            consumer = consumer_cls(self.conf, self.session, topic, callback)
+            self._register_consumer(consumer)
+            return consumer
+
+        return self.ensure(_connect_error, _declare_consumer)
+
+    def iterconsume(self, limit=None, timeout=None):
+        """Return an iterator that will consume from all queues/consumers"""
+
+        def _error_callback(exc):
+            if isinstance(exc, qpid.messaging.exceptions.Empty):
+                LOG.exception(_('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))
+
+        def _consume():
+            nxt_receiver = self.session.next_receiver(timeout=timeout)
+            try:
+                self._lookup_consumer(nxt_receiver).consume()
+            except Exception:
+                LOG.exception(_("Error processing message.  Skipping it."))
+
+        for iteration in itertools.count(0):
+            if limit and iteration >= limit:
+                raise StopIteration
+            yield self.ensure(_error_callback, _consume)
+
+    def cancel_consumer_thread(self):
+        """Cancel a consumer thread"""
+        if self.consumer_thread is not None:
+            self.consumer_thread.kill()
+            try:
+                self.consumer_thread.wait()
+            except greenlet.GreenletExit:
+                pass
+            self.consumer_thread = None
+
+    def publisher_send(self, cls, topic, msg):
+        """Send to a publisher based on the publisher class"""
+
+        def _connect_error(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 _publisher_send():
+            publisher = cls(self.conf, self.session, topic)
+            publisher.send(msg)
+
+        return self.ensure(_connect_error, _publisher_send)
+
+    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):
+        """Create a 'topic' consumer."""
+        self.declare_consumer(functools.partial(TopicConsumer,
+                                                name=queue_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):
+        """Send a 'topic' message"""
+        self.publisher_send(TopicPublisher, topic, msg)
+
+    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)
+
+    def consume(self, limit=None):
+        """Consume from all queues/consumers"""
+        it = self.iterconsume(limit=limit)
+        while True:
+            try:
+                it.next()
+            except StopIteration:
+                return
+
+    def consume_in_thread(self):
+        """Consumer from all queues/consumers in a greenthread"""
+        def _consumer_thread():
+            try:
+                self.consume()
+            except greenlet.GreenletExit:
+                return
+        if self.consumer_thread is None:
+            self.consumer_thread = eventlet.spawn(_consumer_thread)
+        return self.consumer_thread
+
+    def create_consumer(self, topic, proxy, fanout=False):
+        """Create a consumer that calls a method in a proxy object"""
+        proxy_cb = rpc_amqp.ProxyCallback(self.conf, proxy,
+                rpc_amqp.get_connection_pool(self.conf, Connection))
+
+        if fanout:
+            consumer = FanoutConsumer(self.conf, self.session, topic, proxy_cb)
+        else:
+            consumer = TopicConsumer(self.conf, self.session, topic, proxy_cb)
+
+        self._register_consumer(consumer)
+
+        return consumer
+
+    def create_worker(self, topic, proxy, pool_name):
+        """Create a worker that calls a method in a proxy object"""
+        proxy_cb = rpc_amqp.ProxyCallback(self.conf, proxy,
+                rpc_amqp.get_connection_pool(self.conf, Connection))
+
+        consumer = TopicConsumer(self.conf, self.session, topic, proxy_cb,
+                                 name=pool_name)
+
+        self._register_consumer(consumer)
+
+        return consumer
+
+
+def create_connection(conf, new=True):
+    """Create a connection"""
+    return rpc_amqp.create_connection(conf, new,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def multicall(conf, context, topic, msg, timeout=None):
+    """Make a call that returns multiple times."""
+    return rpc_amqp.multicall(conf, context, topic, msg, timeout,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def call(conf, context, topic, msg, timeout=None):
+    """Sends a message on a topic and wait for a response."""
+    return rpc_amqp.call(conf, context, topic, msg, timeout,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def cast(conf, context, topic, msg):
+    """Sends a message on a topic without waiting for a response."""
+    return rpc_amqp.cast(conf, context, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def fanout_cast(conf, context, topic, msg):
+    """Sends a message on a fanout exchange without waiting for a response."""
+    return rpc_amqp.fanout_cast(conf, context, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def cast_to_server(conf, context, server_params, topic, msg):
+    """Sends a message on a topic to a specific server."""
+    return rpc_amqp.cast_to_server(conf, context, server_params, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def fanout_cast_to_server(conf, context, server_params, topic, msg):
+    """Sends a message on a fanout exchange to a specific server."""
+    return rpc_amqp.fanout_cast_to_server(conf, context, server_params, topic,
+            msg, rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def notify(conf, context, topic, msg):
+    """Sends a notification event on a topic."""
+    return rpc_amqp.notify(conf, context, topic, msg,
+            rpc_amqp.get_connection_pool(conf, Connection))
+
+
+def cleanup():
+    return rpc_amqp.cleanup(Connection.pool)
diff --git a/openstack/common/rpc/matchmaker.py b/openstack/common/rpc/matchmaker.py
new file mode 100644
index 0000000..4da1dcd
--- /dev/null
+++ b/openstack/common/rpc/matchmaker.py
@@ -0,0 +1,257 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+#    Copyright 2011 Cloudscaling Group, 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.
+"""
+The MatchMaker classes should except a Topic or Fanout exchange key and
+return keys for direct exchanges, per (approximate) AMQP parlance.
+"""
+
+import contextlib
+import itertools
+import json
+import logging
+
+from openstack.common import cfg
+
+
+matchmaker_opts = [
+    # Matchmaker ring file
+    cfg.StrOpt('matchmaker_ringfile',
+        default='/etc/nova/matchmaker_ring.json',
+        help='Matchmaker ring file (JSON)'),
+]
+
+CONF = cfg.CONF
+CONF.register_opts(matchmaker_opts)
+LOG = logging.getLogger(__name__)
+contextmanager = contextlib.contextmanager
+
+
+class MatchMakerException(Exception):
+    """Signified a match could not be found."""
+    message = _("Match not found by MatchMaker.")
+
+
+class Exchange(object):
+    """
+    Implements lookups.
+    Subclass this to support hashtables, dns, etc.
+    """
+    def __init__(self):
+        pass
+
+    def run(self, key):
+        raise NotImplementedError()
+
+
+class Binding(object):
+    """
+    A binding on which to perform a lookup.
+    """
+    def __init__(self):
+        pass
+
+    def test(self, key):
+        raise NotImplementedError()
+
+
+class MatchMakerBase(object):
+    """Match Maker Base Class."""
+
+    def __init__(self):
+        # Array of tuples. Index [2] toggles negation, [3] is last-if-true
+        self.bindings = []
+
+    def add_binding(self, binding, rule, last=True):
+        self.bindings.append((binding, rule, False, last))
+
+    #NOTE(ewindisch): kept the following method in case we implement the
+    #                 underlying support.
+    #def add_negate_binding(self, binding, rule, last=True):
+    #    self.bindings.append((binding, rule, True, last))
+
+    def queues(self, key):
+        workers = []
+
+        # bit is for negate bindings - if we choose to implement it.
+        # last stops processing rules if this matches.
+        for (binding, exchange, bit, last) in self.bindings:
+            if binding.test(key):
+                workers.extend(exchange.run(key))
+
+                # Support last.
+                if last:
+                    return workers
+        return workers
+
+
+class DirectBinding(Binding):
+    """
+    Specifies a host in the key via a '.' character
+    Although dots are used in the key, the behavior here is
+    that it maps directly to a host, thus direct.
+    """
+    def test(self, key):
+        if '.' in key:
+            return True
+        return False
+
+
+class TopicBinding(Binding):
+    """
+    Where a 'bare' key without dots.
+    AMQP generally considers topic exchanges to be those *with* dots,
+    but we deviate here in terminology as the behavior here matches
+    that of a topic exchange (whereas where there are dots, behavior
+    matches that of a direct exchange.
+    """
+    def test(self, key):
+        if '.' not in key:
+            return True
+        return False
+
+
+class FanoutBinding(Binding):
+    """Match on fanout keys, where key starts with 'fanout.' string."""
+    def test(self, key):
+        if key.startswith('fanout~'):
+            return True
+        return False
+
+
+class StubExchange(Exchange):
+    """Exchange that does nothing."""
+    def run(self, key):
+        return [(key, None)]
+
+
+class RingExchange(Exchange):
+    """
+    Match Maker where hosts are loaded from a static file containing
+    a hashmap (JSON formatted).
+
+    __init__ takes optional ring dictionary argument, otherwise
+    loads the ringfile from CONF.mathcmaker_ringfile.
+    """
+    def __init__(self, ring=None):
+        super(RingExchange, self).__init__()
+
+        if ring:
+            self.ring = ring
+        else:
+            fh = open(CONF.matchmaker_ringfile, 'r')
+            self.ring = json.load(fh)
+            fh.close()
+
+        self.ring0 = {}
+        for k in self.ring.keys():
+            self.ring0[k] = itertools.cycle(self.ring[k])
+
+    def _ring_has(self, key):
+        if key in self.ring0:
+            return True
+        return False
+
+
+class RoundRobinRingExchange(RingExchange):
+    """A Topic Exchange based on a hashmap."""
+    def __init__(self, ring=None):
+        super(RoundRobinRingExchange, self).__init__(ring)
+
+    def run(self, key):
+        if not self._ring_has(key):
+            LOG.warn(
+                _("No key defining hosts for topic '%s', "
+                  "see ringfile") % (key, )
+            )
+            return []
+        host = next(self.ring0[key])
+        return [(key + '.' + host, host)]
+
+
+class FanoutRingExchange(RingExchange):
+    """Fanout Exchange based on a hashmap."""
+    def __init__(self, ring=None):
+        super(FanoutRingExchange, self).__init__(ring)
+
+    def run(self, key):
+        # Assume starts with "fanout~", strip it for lookup.
+        nkey = key.split('fanout~')[1:][0]
+        if not self._ring_has(nkey):
+            LOG.warn(
+                _("No key defining hosts for topic '%s', "
+                  "see ringfile") % (nkey, )
+            )
+            return []
+        return map(lambda x: (key + '.' + x, x), self.ring[nkey])
+
+
+class LocalhostExchange(Exchange):
+    """Exchange where all direct topics are local."""
+    def __init__(self):
+        super(Exchange, self).__init__()
+
+    def run(self, key):
+        return [(key.split('.')[0] + '.localhost', 'localhost')]
+
+
+class DirectExchange(Exchange):
+    """
+    Exchange where all topic keys are split, sending to second half.
+    i.e. "compute.host" sends a message to "compute" running on "host"
+    """
+    def __init__(self):
+        super(Exchange, self).__init__()
+
+    def run(self, key):
+        b, e = key.split('.', 1)
+        return [(b, e)]
+
+
+class MatchMakerRing(MatchMakerBase):
+    """
+    Match Maker where hosts are loaded from a static hashmap.
+    """
+    def __init__(self, ring=None):
+        super(MatchMakerRing, self).__init__()
+        self.add_binding(FanoutBinding(), FanoutRingExchange(ring))
+        self.add_binding(DirectBinding(), DirectExchange())
+        self.add_binding(TopicBinding(), RoundRobinRingExchange(ring))
+
+
+class MatchMakerLocalhost(MatchMakerBase):
+    """
+    Match Maker where all bare topics resolve to localhost.
+    Useful for testing.
+    """
+    def __init__(self):
+        super(MatchMakerLocalhost, self).__init__()
+        self.add_binding(FanoutBinding(), LocalhostExchange())
+        self.add_binding(DirectBinding(), DirectExchange())
+        self.add_binding(TopicBinding(), LocalhostExchange())
+
+
+class MatchMakerStub(MatchMakerBase):
+    """
+    Match Maker where topics are untouched.
+    Useful for testing, or for AMQP/brokered queues.
+    Will not work where knowledge of hosts is known (i.e. zeromq)
+    """
+    def __init__(self):
+        super(MatchMakerLocalhost, self).__init__()
+
+        self.add_binding(FanoutBinding(), StubExchange())
+        self.add_binding(DirectBinding(), StubExchange())
+        self.add_binding(TopicBinding(), StubExchange())
diff --git a/openstack/common/rpc/proxy.py b/openstack/common/rpc/proxy.py
new file mode 100644
index 0000000..4f4dff5
--- /dev/null
+++ b/openstack/common/rpc/proxy.py
@@ -0,0 +1,161 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2012 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""
+A helper class for proxy objects to remote APIs.
+
+For more information about rpc API version numbers, see:
+    rpc/dispatcher.py
+"""
+
+
+from openstack.common import rpc
+
+
+class RpcProxy(object):
+    """A helper class for rpc clients.
+
+    This class is a wrapper around the RPC client API.  It allows you to
+    specify the topic and API version in a single place.  This is intended to
+    be used as a base class for a class that implements the client side of an
+    rpc API.
+    """
+
+    def __init__(self, topic, default_version):
+        """Initialize an RpcProxy.
+
+        :param topic: The topic to use for all messages.
+        :param default_version: The default API version to request in all
+               outgoing messages.  This can be overridden on a per-message
+               basis.
+        """
+        self.topic = topic
+        self.default_version = default_version
+        super(RpcProxy, self).__init__()
+
+    def _set_version(self, msg, vers):
+        """Helper method to set the version in a message.
+
+        :param msg: The message having a version added to it.
+        :param vers: The version number to add to the message.
+        """
+        msg['version'] = vers if vers else self.default_version
+
+    def _get_topic(self, topic):
+        """Return the topic to use for a message."""
+        return topic if topic else self.topic
+
+    @staticmethod
+    def make_msg(method, **kwargs):
+        return {'method': method, 'args': kwargs}
+
+    def call(self, context, msg, topic=None, version=None, timeout=None):
+        """rpc.call() a remote method.
+
+        :param context: The request context
+        :param msg: The message to send, including the method and args.
+        :param topic: Override the topic for this message.
+        :param timeout: (Optional) A timeout to use when waiting for the
+               response.  If no timeout is specified, a default timeout will be
+               used that is usually sufficient.
+        :param version: (Optional) Override the requested API version in this
+               message.
+
+        :returns: The return value from the remote method.
+        """
+        self._set_version(msg, version)
+        return rpc.call(context, self._get_topic(topic), msg, timeout)
+
+    def multicall(self, context, msg, topic=None, version=None, timeout=None):
+        """rpc.multicall() a remote method.
+
+        :param context: The request context
+        :param msg: The message to send, including the method and args.
+        :param topic: Override the topic for this message.
+        :param timeout: (Optional) A timeout to use when waiting for the
+               response.  If no timeout is specified, a default timeout will be
+               used that is usually sufficient.
+        :param version: (Optional) Override the requested API version in this
+               message.
+
+        :returns: An iterator that lets you process each of the returned values
+                  from the remote method as they arrive.
+        """
+        self._set_version(msg, version)
+        return rpc.multicall(context, self._get_topic(topic), msg, timeout)
+
+    def cast(self, context, msg, topic=None, version=None):
+        """rpc.cast() a remote method.
+
+        :param context: The request context
+        :param msg: The message to send, including the method and args.
+        :param topic: Override the topic for this message.
+        :param version: (Optional) Override the requested API version in this
+               message.
+
+        :returns: None.  rpc.cast() does not wait on any return value from the
+                  remote method.
+        """
+        self._set_version(msg, version)
+        rpc.cast(context, self._get_topic(topic), msg)
+
+    def fanout_cast(self, context, msg, version=None):
+        """rpc.fanout_cast() a remote method.
+
+        :param context: The request context
+        :param msg: The message to send, including the method and args.
+        :param version: (Optional) Override the requested API version in this
+               message.
+
+        :returns: None.  rpc.fanout_cast() does not wait on any return value
+                  from the remote method.
+        """
+        self._set_version(msg, version)
+        rpc.fanout_cast(context, self.topic, msg)
+
+    def cast_to_server(self, context, server_params, msg, topic=None,
+            version=None):
+        """rpc.cast_to_server() a remote method.
+
+        :param context: The request context
+        :param server_params: Server parameters.  See rpc.cast_to_server() for
+               details.
+        :param msg: The message to send, including the method and args.
+        :param topic: Override the topic for this message.
+        :param version: (Optional) Override the requested API version in this
+               message.
+
+        :returns: None.  rpc.cast_to_server() does not wait on any
+                  return values.
+        """
+        self._set_version(msg, version)
+        rpc.cast_to_server(context, server_params, self._get_topic(topic), msg)
+
+    def fanout_cast_to_server(self, context, server_params, msg, version=None):
+        """rpc.fanout_cast_to_server() a remote method.
+
+        :param context: The request context
+        :param server_params: Server parameters.  See rpc.cast_to_server() for
+               details.
+        :param msg: The message to send, including the method and args.
+        :param version: (Optional) Override the requested API version in this
+               message.
+
+        :returns: None.  rpc.fanout_cast_to_server() does not wait on any
+                  return values.
+        """
+        self._set_version(msg, version)
+        rpc.fanout_cast_to_server(context, server_params, self.topic, msg)