openstack/deb-python-taskflow
Code Issues Proposed changes

Merge "Rename uuid to topic"

Browse Source
This commit is contained in:
Jenkins
2014-03-06 06:01:51 +00:00
committed by Gerrit Code Review
parent 33a13484d1 92aa862b08
commit 3e495e44f3
4 changed files with 15 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
taskflow/engines/worker_based/proxy.py
View File

@@ -35,9 +35,9 @@ class Proxy(object):
callback when new message received and is used to publish messages.
"""
def __init__(self, uuid, exchange_name, on_message, on_wait=None,
def __init__(self, topic, exchange_name, on_message, on_wait=None,
**kwargs):
self._uuid = uuid
self._topic = topic
self._exchange_name = exchange_name
self._on_message = on_message
self._on_wait = on_wait
@@ -87,7 +87,7 @@ class Proxy(object):
LOG.info("Starting to consume from the '%s' exchange." %
self._exchange_name)
with kombu.connections[self._conn].acquire(block=True) as conn:
queue = self._make_queue(self._uuid, self._exchange, channel=conn)
queue = self._make_queue(self._topic, self._exchange, channel=conn)
try:
with conn.Consumer(queues=queue,
callbacks=[self._on_message]):

4
taskflow/engines/worker_based/server.py
View File

@@ -30,8 +30,8 @@ LOG = logging.getLogger(__name__)
class Server(object):
"""Server implementation that waits for incoming tasks requests."""
def __init__(self, uuid, exchange, executor, endpoints, **kwargs):
self._proxy = proxy.Proxy(uuid, exchange, self._on_message, **kwargs)
def __init__(self, topic, exchange, executor, endpoints, **kwargs):
self._proxy = proxy.Proxy(topic, exchange, self._on_message, **kwargs)
self._executor = executor
self._endpoints = dict([(endpoint.name, endpoint)
for endpoint in endpoints])

12
taskflow/tests/unit/worker_based/test_proxy.py
View File

@@ -28,7 +28,7 @@ class TestProxy(test.MockTestCase):
def setUp(self):
super(TestProxy, self).setUp()
self.uuid = 'test-uuid'
self.topic = 'test-topic'
self.broker_url = 'test-url'
self.exchange_name = 'test-exchange'
self.timeout = 5
@@ -77,14 +77,14 @@ class TestProxy(test.MockTestCase):
# reset master mock
self._reset_master_mock()
def _queue_name(self, uuid):
return "%s_%s" % (self.exchange_name, uuid)
def _queue_name(self, topic):
return "%s_%s" % (self.exchange_name, topic)
def proxy_start_calls(self, calls, exc_type=mock.ANY):
return [
mock.call.Queue(name=self._queue_name(self.uuid),
mock.call.Queue(name=self._queue_name(self.topic),
exchange=self.exchange_inst_mock,
routing_key=self.uuid,
routing_key=self.topic,
durable=False,
queue_arguments=self.queue_arguments,
channel=self.conn_inst_mock),
@@ -99,7 +99,7 @@ class TestProxy(test.MockTestCase):
]
def proxy(self, reset_master_mock=False, **kwargs):
proxy_kwargs = dict(uuid=self.uuid,
proxy_kwargs = dict(topic=self.topic,
exchange_name=self.exchange_name,
on_message=self.on_message_mock,
url=self.broker_url)

8
taskflow/tests/unit/worker_based/test_server.py
View File

@@ -31,7 +31,7 @@ class TestServer(test.MockTestCase):
def setUp(self):
super(TestServer, self).setUp()
self.server_uuid = 'server-uuid'
self.server_topic = 'server-topic'
self.server_exchange = 'server-exchange'
self.broker_url = 'test-url'
self.task_uuid = 'task-uuid'
@@ -59,7 +59,7 @@ class TestServer(test.MockTestCase):
self.master_mock.attach_mock(self.message_mock, 'message')
def server(self, reset_master_mock=False, **kwargs):
server_kwargs = dict(uuid=self.server_uuid,
server_kwargs = dict(topic=self.server_topic,
exchange=self.server_exchange,
executor=self.executor_mock,
endpoints=self.endpoints,
@@ -98,7 +98,7 @@ class TestServer(test.MockTestCase):
# check calls
master_mock_calls = [
mock.call.Proxy(self.server_uuid, self.server_exchange,
mock.call.Proxy(self.server_topic, self.server_exchange,
s._on_message, url=self.broker_url)
]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
@@ -109,7 +109,7 @@ class TestServer(test.MockTestCase):
# check calls
master_mock_calls = [
mock.call.Proxy(self.server_uuid, self.server_exchange,
mock.call.Proxy(self.server_topic, self.server_exchange,
s._on_message, url=self.broker_url)
]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls)