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

1 files changed, 184 insertions, 0 deletions

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