openstack
/
qinling
Code Issues Proposed changes

Support sync/async execution 

Invoke a function in sync way means the command will wait until the
execution finish. On the contrary, 'sync=false' means user will get
a response before the function is actually invoked. User could get
execution status using execution ID in the response.

Change-Id: I605ea8414f49d44899e0bc8c36b89e48e335d284
This commit is contained in:
Lingxian Kong 2017-05-18 22:38:20 +12:00
parent 14789ab0c8
commit d80f3acf35
2 changed files with 27 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
qinling/api/controllers/v1/execution.py
View File

@ -45,6 +45,7 @@ class ExecutionsController(rest.RestController):
LOG.info("Creating execution. [execution=%s]", params) LOG.info("Creating execution. [execution=%s]", params)
function_id = params['function_id'] function_id = params['function_id']
is_sync = params.get('sync', True)
# Check if the service url is existing. # Check if the service url is existing.
try: try:
@ -71,12 +72,14 @@ class ExecutionsController(rest.RestController):
db_model = db_api.create_execution(params) db_model = db_api.create_execution(params)
self.engine_client.create_execution( self.engine_client.create_execution(
db_model.id, function_id, runtime_id, input=params.get('input') db_model.id, function_id, runtime_id, input=params.get('input'),
is_sync=is_sync
) )
updated_db = db_api.get_execution(db_model.id) if is_sync:
db_model = db_api.get_execution(db_model.id)
return resources.Execution.from_dict(updated_db.to_dict()) return resources.Execution.from_dict(db_model.to_dict())
@rest_utils.wrap_wsme_controller_exception @rest_utils.wrap_wsme_controller_exception
@wsme_pecan.wsexpose(resources.Executions) @wsme_pecan.wsexpose(resources.Executions)

30
qinling/rpc.py
View File

@ -147,15 +147,27 @@ class EngineClient(object):
@wrap_messaging_exception @wrap_messaging_exception
def create_execution(self, execution_id, function_id, runtime_id, def create_execution(self, execution_id, function_id, runtime_id,
input=None): input=None, is_sync=True):
return self._client.prepare(topic=self.topic, server=None).call( method_client = self._client.prepare(topic=self.topic, server=None)
ctx.get_ctx(),
'create_execution', if is_sync:
execution_id=execution_id, return method_client.call(
function_id=function_id, ctx.get_ctx(),
runtime_id=runtime_id, 'create_execution',
input=input execution_id=execution_id,
) function_id=function_id,
runtime_id=runtime_id,
input=input
)
else:
method_client.cast(
ctx.get_ctx(),
'create_execution',
execution_id=execution_id,
function_id=function_id,
runtime_id=runtime_id,
input=input
)
@wrap_messaging_exception @wrap_messaging_exception
def delete_function(self, id, name): def delete_function(self, id, name):