98 support router tasks (#217)

* support router tasks

Signed-off-by: Jack Schofield <[email protected]>

Author: jackschofield23

src/Contracts/Constants/TaskTypeConstants.cs

@@ -0,0 +1,7 @@
+namespace Monai.Deploy.WorkflowManager.Contracts.Constants
+{
+    public static class TaskTypeConstants
+    {
+        public const string RouterTask = "router";
+    }
+}

src/WorkflowExecuter/Services/WorkflowExecuterService.cs

@@ -13,6 +13,7 @@
 using Monai.Deploy.WorkflowManager.Common.Interfaces;
 using Monai.Deploy.WorkflowManager.ConditionsResolver.Parser;
 using Monai.Deploy.WorkflowManager.Configuration;
+using Monai.Deploy.WorkflowManager.Contracts.Constants;
 using Monai.Deploy.WorkflowManager.Contracts.Models;
 using Monai.Deploy.WorkflowManager.Database.Interfaces;
 using Monai.Deploy.WorkflowManager.Logging.Logging;
@@ -125,6 +126,15 @@ await _workflowRepository.GetWorkflowsByAeTitleAsync(message.CalledAeTitle) as L
                     continue;
                 }
 
+                if (string.Equals(task.TaskType, TaskTypeConstants.RouterTask, StringComparison.InvariantCultureIgnoreCase))
+                {
+                    var workflow = await _workflowRepository.GetByWorkflowIdAsync(workflowInstance.WorkflowId);
+
+                    await HandleTaskDestinations(workflowInstance, workflow, task, message.CorrelationId);
+
+                    continue;
+                }
+
                 if (task.Status != TaskExecutionStatus.Created)
                 {
                     _logger.TaskPreviouslyDispatched(workflowInstance.PayloadId, task.TaskId);
@@ -213,20 +223,7 @@ public async Task<bool> ProcessTaskUpdate(TaskUpdateEvent message)
                 _logger.ExportFilesNotFound(currentTask.TaskId, workflowInstance.Id);
             }
 
-            var newTaskExecutions = await CreateTaskDestinations(workflowInstance, workflow, message.TaskId);
-
-            if (!newTaskExecutions.Any())
-            {
-                await UpdateWorkflowInstanceStatus(workflowInstance, message.TaskId, message.Status);
-
-                return await CompleteTask(currentTask, workflowInstance, message.CorrelationId, message.Status);
-            }
-
-            var processed = await HandleTaskDestinations(workflowInstance, message.CorrelationId, newTaskExecutions);
-
-            processed &= await CompleteTask(currentTask, workflowInstance, message.CorrelationId, message.Status);
-
-            return processed;
+            return await HandleTaskDestinations(workflowInstance, workflow, currentTask, message.CorrelationId);
         }
 
         public async Task<bool> ProcessExportComplete(ExportCompleteEvent message, string correlationId)
@@ -251,20 +248,7 @@ public async Task<bool> ProcessExportComplete(ExportCompleteEvent message, strin
                     return false;
                 }
 
-                var newTaskExecutions = await CreateTaskDestinations(workflowInstance, workflow, task.TaskId);
-
-                if (!newTaskExecutions.Any())
-                {
-                    await UpdateWorkflowInstanceStatus(workflowInstance, task.TaskId, TaskExecutionStatus.Succeeded);
-
-                    return await CompleteTask(task, workflowInstance, correlationId, TaskExecutionStatus.Succeeded);
-                }
-
-                var processed = await HandleTaskDestinations(workflowInstance, correlationId, newTaskExecutions);
-
-                processed &= await CompleteTask(task, workflowInstance, correlationId, TaskExecutionStatus.Succeeded);
-
-                return processed;
+                return await HandleTaskDestinations(workflowInstance, workflow, task, correlationId);
             }
 
             if ((message.Status.Equals(ExportStatus.Failure) || message.Status.Equals(ExportStatus.PartialFailure)) &&
@@ -349,7 +333,7 @@ private async Task<bool> HandleOutputArtifacts(WorkflowInstance workflowInstance
             return true;
         }
 
-        private async Task<bool> HandleTaskDestinations(WorkflowInstance workflowInstance, string correlationId, IList<TaskExecution> taskExecutions)
+        private async Task<bool> DispatchTaskDestinations(WorkflowInstance workflowInstance, WorkflowRevision workflow, string correlationId, IList<TaskExecution> taskExecutions)
         {
             workflowInstance.Tasks?.AddRange(taskExecutions);
 
@@ -362,9 +346,23 @@ private async Task<bool> HandleTaskDestinations(WorkflowInstance workflowInstanc
 
             foreach (var taskExec in taskExecutions)
             {
+                if (string.Equals(taskExec.TaskType, TaskTypeConstants.RouterTask, StringComparison.InvariantCultureIgnoreCase))
+                {
+                    processed &= await HandleTaskDestinations(workflowInstance, workflow, taskExec, correlationId);
+
+                    if (processed is false)
+                    {
+                        continue;
+                    }
+
+                    await _workflowInstanceRepository.UpdateTaskStatusAsync(workflowInstance.Id, taskExec.TaskId, TaskExecutionStatus.Succeeded);
+
+                    continue;
+                }
+
                 processed &= await DispatchTask(workflowInstance, taskExec, correlationId);
 
-                if (!processed)
+                if (processed is false)
                 {
                     continue;
                 }
@@ -375,6 +373,22 @@ private async Task<bool> HandleTaskDestinations(WorkflowInstance workflowInstanc
             return processed;
         }
 
+        private async Task<bool> HandleTaskDestinations(WorkflowInstance workflowInstance, WorkflowRevision workflow, TaskExecution task, string correlationId)
+        {
+            var newTaskExecutions = await CreateTaskDestinations(workflowInstance, workflow, task.TaskId);
+
+            if (newTaskExecutions.Any() is false)
+            {
+                await UpdateWorkflowInstanceStatus(workflowInstance, task.TaskId, TaskExecutionStatus.Succeeded);
+
+                return await CompleteTask(task, workflowInstance, correlationId, TaskExecutionStatus.Succeeded);
+            }
+
+            await DispatchTaskDestinations(workflowInstance, workflow, correlationId, newTaskExecutions);
+
+            return await CompleteTask(task, workflowInstance, correlationId, TaskExecutionStatus.Succeeded);
+        }
+
         private async Task<List<TaskExecution>> CreateTaskDestinations(WorkflowInstance workflowInstance, WorkflowRevision workflow, string taskId)
         {
             var currentTaskDestinations = workflow.Workflow?.Tasks?.SingleOrDefault(t => t.Id == taskId)?.TaskDestinations;

src/WorkflowManager/appsettings.Development.json

@@ -14,7 +14,9 @@
         "accessToken": "minioadmin",
         "bucket": "test-bucket",
         "region": "eu-west-2",
-        "securedConnection": false
+        "securedConnection": false,
+        "executableLocation": "/.",
+        "serviceName": "serviceName"
       }
     },
     "messaging": {

tests/IntegrationTests/WorkflowExecutor.IntegrationTests/Features/WorkflowRequest.feature

@@ -14,6 +14,17 @@ Scenario Outline: Publish a valid workflow request which creates a single workfl
     | Basic_Workflow_1      | Basic_AeTitle_WF_Request   |
     | Basic_Workflow_1      | Basic_Id_WF_Request        |
 
+@WorkflowRequest
+Scenario Outline: Publish a valid workflow request which creates a single workflow instance with routing task
+    Given I have a clinical workflow <workflow>
+    And I have a bucket in MinIO bucket1
+    When I publish a Workflow Request Message <workflowRequestMessage>
+    Then I can see 1 Workflow Instance is created
+    And 2 Task Dispatch event is published
+    Examples:
+    | workflow              | workflowRequestMessage     |
+    | Routing_Workflow_1    | Routing_Id_WF_Request   |
+
 @WorkflowRequest
 Scenario Outline: Publish a valid workflow request which creates multiple workflow instances
     Given I have a clinical workflow <workflow_1>