openstack/taskflow
Code Issues Proposed changes

Rework proxy publish functionality

Browse Source 
* Task uuid is not a required parameter now;
* Exchange declaration does not require a connection object.

Change-Id: I9a77c9afe7d9fad007902d77d24a0b1a96c6c519
This commit is contained in:
Stanislav Kudriashev
2014-03-11 15:00:26 +02:00
committed by Stanislav Kudriashev
parent 37bdae0f0c
commit 8d3c00ccfd
6 changed files with 41 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
taskflow/engines/worker_based/executor.py
View File

@@ -126,8 +126,9 @@ class WorkerTaskExecutor(executor.TaskExecutorBase):
# publish request # publish request
request = remote_task.request request = remote_task.request
LOG.debug("Sending request: %s", request) LOG.debug("Sending request: %s", request)
self._proxy.publish(request, remote_task.uuid, self._proxy.publish(request, routing_key=topic,
routing_key=topic, reply_to=self._uuid) reply_to=self._uuid,
correlation_id=remote_task.uuid)
except Exception: except Exception:
with misc.capture_failure() as failure: with misc.capture_failure() as failure:
LOG.exception("Failed to submit the '%s' task", remote_task) LOG.exception("Failed to submit the '%s' task", remote_task)

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

@@ -48,7 +48,6 @@ class Proxy(object):
# create exchange # create exchange
self._exchange = kombu.Exchange(name=self._exchange_name, self._exchange = kombu.Exchange(name=self._exchange_name,
channel=self._conn,
durable=False, durable=False,
auto_delete=True) auto_delete=True)
@@ -66,14 +65,13 @@ class Proxy(object):
auto_delete=True, auto_delete=True,
**kwargs) **kwargs)
def publish(self, msg, task_uuid, routing_key, **kwargs): def publish(self, msg, routing_key, **kwargs):
"""Publish message to the named exchange with routing key.""" """Publish message to the named exchange with routing key."""
with kombu.producers[self._conn].acquire(block=True) as producer: with kombu.producers[self._conn].acquire(block=True) as producer:
queue = self._make_queue(routing_key, self._exchange) queue = self._make_queue(routing_key, self._exchange)
producer.publish(body=msg, producer.publish(body=msg,
routing_key=routing_key, routing_key=routing_key,
exchange=self._exchange, exchange=self._exchange,
correlation_id=task_uuid,
declare=[queue], declare=[queue],
**kwargs) **kwargs)

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

@@ -103,7 +103,7 @@ class Server(object):
response = dict(state=state, **kwargs) response = dict(state=state, **kwargs)
LOG.debug("Sending reply: %s", response) LOG.debug("Sending reply: %s", response)
try: try:
self._proxy.publish(response, task_uuid, reply_to) self._proxy.publish(response, reply_to, correlation_id=task_uuid)
except Exception: except Exception:
LOG.exception("Failed to send reply") LOG.exception("Failed to send reply")

15
taskflow/tests/unit/worker_based/test_executor.py
View File

@@ -222,9 +222,10 @@ class TestWorkerTaskExecutor(test.MockTestCase):
result = ex.execute_task(self.task, self.task_uuid, self.task_args) result = ex.execute_task(self.task, self.task_uuid, self.task_args)
expected_calls = [ expected_calls = [
mock.call.proxy.publish(request, self.task_uuid, mock.call.proxy.publish(request,
routing_key=self.executor_topic, routing_key=self.executor_topic,
reply_to=self.executor_uuid) reply_to=self.executor_uuid,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, expected_calls) self.assertEqual(self.master_mock.mock_calls, expected_calls)
self.assertIsInstance(result, futures.Future) self.assertIsInstance(result, futures.Future)
@@ -238,9 +239,10 @@ class TestWorkerTaskExecutor(test.MockTestCase):
self.task_result, self.task_failures) self.task_result, self.task_failures)
expected_calls = [ expected_calls = [
mock.call.proxy.publish(request, self.task_uuid, mock.call.proxy.publish(request,
routing_key=self.executor_topic, routing_key=self.executor_topic,
reply_to=self.executor_uuid) reply_to=self.executor_uuid,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, expected_calls) self.assertEqual(self.master_mock.mock_calls, expected_calls)
self.assertIsInstance(result, futures.Future) self.assertIsInstance(result, futures.Future)
@@ -265,9 +267,10 @@ class TestWorkerTaskExecutor(test.MockTestCase):
result = ex.execute_task(self.task, self.task_uuid, self.task_args) result = ex.execute_task(self.task, self.task_uuid, self.task_args)
expected_calls = [ expected_calls = [
mock.call.proxy.publish(request, self.task_uuid, mock.call.proxy.publish(request,
routing_key=self.executor_topic, routing_key=self.executor_topic,
reply_to=self.executor_uuid) reply_to=self.executor_uuid,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, expected_calls) self.assertEqual(self.master_mock.mock_calls, expected_calls)

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

@@ -109,7 +109,6 @@ class TestProxy(test.MockTestCase):
mock.call.Connection(self.broker_url, transport=None, mock.call.Connection(self.broker_url, transport=None,
transport_options=None), transport_options=None),
mock.call.Exchange(name=self.exchange_name, mock.call.Exchange(name=self.exchange_name,
channel=self.conn_inst_mock,
durable=False, durable=False,
auto_delete=True) auto_delete=True)
] ]
@@ -123,7 +122,6 @@ class TestProxy(test.MockTestCase):
mock.call.Connection(self.broker_url, transport='memory', mock.call.Connection(self.broker_url, transport='memory',
transport_options=transport_opts), transport_options=transport_opts),
mock.call.Exchange(name=self.exchange_name, mock.call.Exchange(name=self.exchange_name,
channel=self.conn_inst_mock,
durable=False, durable=False,
auto_delete=True) auto_delete=True)
] ]
@@ -136,7 +134,7 @@ class TestProxy(test.MockTestCase):
kwargs = dict(a='a', b='b') kwargs = dict(a='a', b='b')
self.proxy(reset_master_mock=True).publish( self.proxy(reset_master_mock=True).publish(
task_data, task_uuid, routing_key, **kwargs) task_data, routing_key, correlation_id=task_uuid, **kwargs)
master_mock_calls = [ master_mock_calls = [
mock.call.Queue(name=self._queue_name(routing_key), mock.call.Queue(name=self._queue_name(routing_key),

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

@@ -219,8 +219,8 @@ class TestServer(test.MockTestCase):
s._reply(self.reply_to, self.task_uuid) s._reply(self.reply_to, self.task_uuid)
self.assertEqual(self.master_mock.mock_calls, [ self.assertEqual(self.master_mock.mock_calls, [
mock.call.proxy.publish({'state': 'FAILURE'}, self.task_uuid, mock.call.proxy.publish({'state': 'FAILURE'}, self.reply_to,
self.reply_to) correlation_id=self.task_uuid)
]) ])
self.assertTrue(mocked_exception.called) self.assertTrue(mocked_exception.called)
@@ -234,14 +234,14 @@ class TestServer(test.MockTestCase):
# check calls # check calls
master_mock_calls = [ master_mock_calls = [
mock.call.proxy.publish(self.resp_running, self.task_uuid, mock.call.proxy.publish(self.resp_running, self.reply_to,
self.reply_to), correlation_id=self.task_uuid),
mock.call.proxy.publish(self.resp_progress(0.0), self.task_uuid, mock.call.proxy.publish(self.resp_progress(0.0), self.reply_to,
self.reply_to), correlation_id=self.task_uuid),
mock.call.proxy.publish(self.resp_progress(1.0), self.task_uuid, mock.call.proxy.publish(self.resp_progress(1.0), self.reply_to,
self.reply_to), correlation_id=self.task_uuid),
mock.call.proxy.publish(self.resp_success(5), self.task_uuid, mock.call.proxy.publish(self.resp_success(5), self.reply_to,
self.reply_to) correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls) self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
@@ -252,10 +252,10 @@ class TestServer(test.MockTestCase):
# check calls # check calls
master_mock_calls = [ master_mock_calls = [
mock.call.proxy.publish(self.resp_running, self.task_uuid, mock.call.proxy.publish(self.resp_running, self.reply_to,
self.reply_to), correlation_id=self.task_uuid),
mock.call.proxy.publish(self.resp_success(1), self.task_uuid, mock.call.proxy.publish(self.resp_success(1), self.reply_to,
self.reply_to) correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls) self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
@@ -283,7 +283,8 @@ class TestServer(test.MockTestCase):
# check calls # check calls
master_mock_calls = [ master_mock_calls = [
mock.call.proxy.publish(self.resp_failure(failure_dict), mock.call.proxy.publish(self.resp_failure(failure_dict),
self.task_uuid, self.reply_to) self.reply_to,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls) self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
@@ -300,7 +301,8 @@ class TestServer(test.MockTestCase):
# check calls # check calls
master_mock_calls = [ master_mock_calls = [
mock.call.proxy.publish(self.resp_failure(failure_dict), mock.call.proxy.publish(self.resp_failure(failure_dict),
self.task_uuid, self.reply_to) self.reply_to,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls) self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
@@ -316,10 +318,11 @@ class TestServer(test.MockTestCase):
# check calls # check calls
master_mock_calls = [ master_mock_calls = [
mock.call.proxy.publish(self.resp_running, self.task_uuid, mock.call.proxy.publish(self.resp_running, self.reply_to,
self.reply_to), correlation_id=self.task_uuid),
mock.call.proxy.publish(self.resp_failure(failure_dict), mock.call.proxy.publish(self.resp_failure(failure_dict),
self.task_uuid, self.reply_to) self.reply_to,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls) self.assertEqual(self.master_mock.mock_calls, master_mock_calls)
@@ -336,10 +339,11 @@ class TestServer(test.MockTestCase):
# check calls # check calls
master_mock_calls = [ master_mock_calls = [
mock.call.proxy.publish(self.resp_running, self.task_uuid, mock.call.proxy.publish(self.resp_running, self.reply_to,
self.reply_to), correlation_id=self.task_uuid),
mock.call.proxy.publish(self.resp_failure(failure_dict), mock.call.proxy.publish(self.resp_failure(failure_dict),
self.task_uuid, self.reply_to) self.reply_to,
correlation_id=self.task_uuid)
] ]
self.assertEqual(self.master_mock.mock_calls, master_mock_calls) self.assertEqual(self.master_mock.mock_calls, master_mock_calls)