Task input type

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/helper/DAGWalkHelper.java

@@ -26,7 +26,6 @@
 import com.weibo.rill.flow.olympicene.core.model.dag.DAGInfo;
 import com.weibo.rill.flow.olympicene.core.model.dag.DAGStatus;
 import com.weibo.rill.flow.olympicene.core.model.strategy.CallbackConfig;
-import com.weibo.rill.flow.interfaces.model.task.FunctionTask;
 import com.weibo.rill.flow.olympicene.core.model.task.TaskCategory;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
@@ -53,22 +52,82 @@ public static DAGWalkHelper getInstance() {
     }
 
     public Set<TaskInfo> getReadyToRunTasks(Collection<TaskInfo> taskInfos) {
-        Set<TaskInfo> readyToRunTasks = taskInfos.stream()
-                .filter(taskInfo -> taskInfo != null && taskInfo.getTaskStatus() == TaskStatus.NOT_STARTED)
-                .filter(taskInfo -> !taskInfo.getTask().isKeyCallback())
-                .filter(taskInfo -> CollectionUtils.isEmpty(taskInfo.getDependencies()) || taskInfo.getDependencies().stream().allMatch(i -> i.getTaskStatus().isSuccessOrSkip()))
+        boolean isKeyMode = isKeyMode(taskInfos);
+        // 根据依赖关系获取准备运行的非流式输入任务
+        Set<TaskInfo> readyToRunTasks = getReadyToRunBlockInputTasks(taskInfos, isKeyMode);
+        // 添加准备运行的流式输入任务
+        addReadyToRunStreamInputTasks(taskInfos, readyToRunTasks, isKeyMode);
+        return readyToRunTasks;
+    }
+
+    /**
+     * 筛选出准备运行的任务:
+     * 1. 当前任务不为空且状态为未开始
+     * 2. 依赖任务全部完成（如果在关键路径模式下，则包括关键路径完成）
+     */
+    private Set<TaskInfo> getReadyToRunBlockInputTasks(Collection<TaskInfo> taskInfos, boolean isKeyMode) {
+        return taskInfos.stream()
+                .filter(Objects::nonNull)
+                .filter(taskInfo -> taskInfo.getTaskStatus() == TaskStatus.NOT_STARTED)
+                .filter(taskInfo -> TaskInputOutputType.getTypeByValue(taskInfo.getTask().getInputType()) == TaskInputOutputType.BLOCK)
+                .filter(taskInfo -> isDependenciesAllSuccessOrSkip(taskInfo, isKeyMode))
                 .collect(Collectors.toSet());
+    }
 
-        if (isKeyMode(taskInfos)) {
-            Set<TaskInfo> keyCallbackTasks = taskInfos.stream()
-                    .filter(taskInfo -> taskInfo != null && taskInfo.getTaskStatus() == TaskStatus.NOT_STARTED)
-                    .filter(taskInfo -> taskInfo.getTask().isKeyCallback()) // 此类型任务只需前置依赖节点关键路径完成即可执行
-                    .filter(taskInfo -> CollectionUtils.isEmpty(taskInfo.getDependencies()) || taskInfo.getDependencies().stream().allMatch(i -> i.getTaskStatus().isSuccessOrKeySuccessOrSkip()))
-                    .collect(Collectors.toSet());
-            readyToRunTasks.addAll(keyCallbackTasks);
+    /**
+     * 非流式输入任务，所有依赖的 block 输出任务都已经完成，或在关键路径模式下关键路径完成或跳过，则任务可以运行
+     */
+    private boolean isDependenciesAllSuccessOrSkip(TaskInfo taskInfo, boolean isKeyMode) {
+        // 1. 没有依赖视为依赖均已完成
+        if (CollectionUtils.isEmpty(taskInfo.getDependencies())) {
+            return true;
         }
+        boolean isTaskKeyMode = isKeyMode && taskInfo.getTask().isKeyCallback();
+        // 2. 非流式输入任务，所有依赖任务是否都已完成（非关键路径模式下完成或跳过，或者在关键路径模式下关键路径完成或跳过）
+        return taskInfo.getDependencies().stream().allMatch(dependency -> isTaskSuccessOrSkip(dependency, isTaskKeyMode));
+    }
 
-        return readyToRunTasks;
+    private boolean isTaskSuccessOrSkip(TaskInfo taskInfo, boolean isTaskKeyMode) {
+        return taskInfo.getTaskStatus().isSuccessOrSkip()
+                || (isTaskKeyMode && taskInfo.getTaskStatus().isSuccessOrKeySuccessOrSkip());
+    }
+
+    /**
+     * 获取准备运行的流式输入任务，并添加到 readyToRunTasks
+     * 
+     * @param taskInfos 所有任务的集合
+     * @param readyToRunTasks 已准备运行的任务集合
+     */
+    private void addReadyToRunStreamInputTasks(Collection<TaskInfo> taskInfos, Set<TaskInfo> readyToRunTasks, boolean isKeyMode) {
+        Set<TaskInfo> readyToRunStreamTasks = new HashSet<>();
+        taskInfos.stream().filter(Objects::nonNull).filter(taskInfo -> taskInfo.getTaskStatus() == TaskStatus.NOT_STARTED)
+                .filter(taskInfo -> TaskInputOutputType.getTypeByValue(taskInfo.getTask().getInputType()) == TaskInputOutputType.STREAM)
+                .forEach(taskInfo -> {
+                    boolean isTaskKeyMode = isKeyMode && taskInfo.getTask().isKeyCallback();
+                    boolean needRun = taskInfo.getDependencies().stream()
+                            .anyMatch(dependency -> isStreamTaskReadyToRun(readyToRunTasks, dependency, isTaskKeyMode));
+                    if (needRun) {
+                        readyToRunStreamTasks.add(taskInfo);
+                    }
+                });
+        readyToRunTasks.addAll(readyToRunStreamTasks);
+    }
+
+    /**
+     * 判断流式输入任务是否可执行，只要有任何依赖符合以下任一条件，流式输入任务就可以执行
+     * 1. 任意依赖的流式输出任务开始执行或者准备被执行
+     * 2. 任意依赖的非流式输出任务执行完成（包括关键路径模式下关键路径执行完成）
+     */
+    private boolean isStreamTaskReadyToRun(Set<TaskInfo> readyToRunTasks, TaskInfo dependency, boolean isTaskKeyMode) {
+        TaskInputOutputType dependencyOutputType = TaskInputOutputType.getTypeByValue(dependency.getTask().getOutputType());
+        boolean streamDependencyStarted = false;
+        boolean blockDependencyFinished = false;
+        if (dependencyOutputType == TaskInputOutputType.STREAM) {
+            streamDependencyStarted = dependency.getTaskStatus() != TaskStatus.NOT_STARTED || readyToRunTasks.contains(dependency);
+        } else {
+            blockDependencyFinished = isTaskSuccessOrSkip(dependency, isTaskKeyMode);
+        }
+        return streamDependencyStarted || blockDependencyFinished;
     }
 
     private boolean isKeyMode(Collection<TaskInfo> allTasks) {

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/dag/DAG.java

@@ -16,6 +16,7 @@
 
 package com.weibo.rill.flow.olympicene.core.model.dag;
 
+import com.fasterxml.jackson.annotation.JsonAlias;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.weibo.rill.flow.interfaces.model.mapping.Mapping;
@@ -43,6 +44,7 @@ public class DAG {
     private List<BaseTask> tasks;
     @Setter
     private List<BaseResource> resources;
+    @JsonProperty("callback")
     private CallbackConfig callbackConfig;
     @Setter
     private Map<String, String> defaultContext;
@@ -61,7 +63,7 @@ public DAG(@JsonProperty("workspace") String workspace,
                @JsonProperty("timer") Timeline timeline,
                @JsonProperty("tasks") List<BaseTask> tasks,
                @JsonProperty("resources") List<BaseResource> resources,
-               @JsonProperty("callback") CallbackConfig callbackConfig,
+               @JsonProperty("callback") @JsonAlias({"callback_config", "callbackConfig"}) CallbackConfig callbackConfig,
                @JsonProperty("defaultContext") Map<String, String> defaultContext,
                @JsonProperty("commonMapping") Map<String, List<Mapping>> commonMapping,
                @JsonProperty("namespace") String namespace,

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/ChoiceTask.java

@@ -52,12 +52,14 @@
             @JsonProperty("progress") Progress progress,
             @JsonProperty("degrade") Degrade degrade,
             @JsonProperty("timeline") Timeline timeline,
-            @JsonProperty("isKeyCallback") boolean isKeyCallback,
+            @JsonProperty("keyCallback") boolean keyCallback,
             @JsonProperty("keyExp") String keyExp,
             @JsonProperty("parameters") Map<String, Object> parameters,
-            @JsonProperty("templateId") String templateId) {
-        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
-                timeline, isKeyCallback, keyExp, parameters, templateId, input);
+            @JsonProperty("templateId") String templateId,
+            @JsonProperty("inputType") String inputType,
+            @JsonProperty("outputType") String outputType) {
+        super(name, title, description, category, next, false, inputMappings, outputMappings, progress,
+                degrade, timeline, keyCallback, keyExp, parameters, templateId, input, inputType, outputType);
         this.choices = choices;
     }
 

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/ForeachTask.java

@@ -55,13 +55,15 @@ public ForeachTask(
             @JsonProperty("progress") Progress progress,
             @JsonProperty("degrade") Degrade degrade,
             @JsonProperty("timeline") Timeline timeline,
-            @JsonProperty("isKeyCallback") boolean isKeyCallback,
+            @JsonProperty("keyCallback") boolean keyCallback,
             @JsonProperty("keyExp") String keyExp,
             @JsonProperty("parameters") Map<String, Object> parameters,
             @JsonProperty("templateId") String templateId,
-            @JsonProperty("input") Map<String, Object> input) {
+            @JsonProperty("input") Map<String, Object> input,
+            @JsonProperty("inputType") String inputType,
+            @JsonProperty("outputType") String outputType) {
         super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
-                timeline, isKeyCallback, keyExp, parameters, templateId, input);
+                timeline, keyCallback, keyExp, parameters, templateId, input, inputType, outputType);
         this.synchronization = synchronization;
         this.iterationMapping = iterationMapping;
         this.tasks = tasks;

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/PassTask.java

@@ -50,13 +50,15 @@
                     @JsonProperty("progress") Progress progress,
                     @JsonProperty("degrade") Degrade degrade,
                     @JsonProperty("timeline") Timeline timeline,
-                    @JsonProperty("isKeyCallback") boolean isKeyCallback,
+                    @JsonProperty("keyCallback") boolean keyCallback,
                     @JsonProperty("keyExp") String keyExp,
                     @JsonProperty("parameters") Map<String, Object> parameters,
                     @JsonProperty("templateId") String templateId,
-                    @JsonProperty("input") Map<String, Object> input) {
-        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
-                timeline, isKeyCallback, keyExp, parameters, templateId, input);
+                    @JsonProperty("input") Map<String, Object> input,
+                    @JsonProperty("inputType") String inputType,
+                    @JsonProperty("outputType") String outputType) {
+        super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade, timeline,
+                keyCallback, keyExp, parameters, templateId, input, inputType, outputType);
         Optional.ofNullable(timeline).ifPresent(it -> it.setTimeoutInSeconds(null));
     }
 

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/ReturnTask.java

@@ -49,13 +49,15 @@ public ReturnTask(@JsonProperty("name") String name,
                       @JsonProperty("progress") Progress progress,
                       @JsonProperty("degrade") Degrade degrade,
                       @JsonProperty("timeline") Timeline timeline,
-                      @JsonProperty("isKeyCallback") boolean isKeyCallback,
+                      @JsonProperty("keyCallback") boolean keyCallback,
                       @JsonProperty("keyExp") String keyExp,
                       @JsonProperty("parameters") Map<String, Object> parameters,
                       @JsonProperty("templateId") String templateId,
-                      @JsonProperty("input") Map<String, Object> input) {
+                      @JsonProperty("input") Map<String, Object> input,
+                      @JsonProperty("inputType") String inputType,
+                      @JsonProperty("outputType") String outputType) {
         super(name, title, description, category, next, false, inputMappings, outputMappings, progress, degrade,
-                timeline, isKeyCallback, keyExp, parameters, templateId, input);
+                timeline, keyCallback, keyExp, parameters, templateId, input, inputType, outputType);
         Optional.ofNullable(timeline).ifPresent(it -> it.setTimeoutInSeconds(null));
         this.conditions = conditions;
     }

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/SuspenseTask.java

@@ -52,13 +52,15 @@ public SuspenseTask(@JsonProperty("name") String name,
                         @JsonProperty("progress") Progress progress,
                         @JsonProperty("degrade") Degrade degrade,
                         @JsonProperty("timeline") Timeline timeline,
-                        @JsonProperty("isKeyCallback") boolean isKeyCallback,
+                        @JsonProperty("keyCallback") boolean keyCallback,
                         @JsonProperty("keyExp") String keyExp,
                         @JsonProperty("parameters") Map<String, Object> parameters,
                         @JsonProperty("templateId") String templateId,
-                        @JsonProperty("input") Map<String, Object> input) {
+                        @JsonProperty("input") Map<String, Object> input,
+                        @JsonProperty("inputType") String inputType,
+                        @JsonProperty("outputType") String outputType) {
         super(name, title, description, category, next, tolerance, inputMappings, outputMappings, progress, degrade,
-                timeline, isKeyCallback, keyExp, parameters, templateId, input);
+                timeline, keyCallback, keyExp, parameters, templateId, input, inputType, outputType);
         this.conditions = conditions;
         this.interruptions = interruptions;
     }

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/SwitchTask.java

@@ -27,7 +27,9 @@
 import lombok.Getter;
 import lombok.Setter;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 @Getter
@@ -48,13 +50,15 @@ public SwitchTask(
             @JsonProperty("progress") Progress progress,
             @JsonProperty("degrade") Degrade degrade,
             @JsonProperty("timeline") Timeline timeline,
-            @JsonProperty("isKeyCallback") boolean isKeyCallback,
+            @JsonProperty("keyCallback") boolean keyCallback,
             @JsonProperty("keyExp") String keyExp,
             @JsonProperty("parameters") Map<String, Object> parameters,
             @JsonProperty("templateId") String templateId,
-            @JsonProperty("input") Map<String, Object> input) {
+            @JsonProperty("input") Map<String, Object> input,
+            @JsonProperty("inputType") String inputType,
+            @JsonProperty("outputType") String outputType) {
         super(name, title, description, category, null, false, inputMappings, null, progress,
-                degrade, timeline, isKeyCallback, keyExp, parameters, templateId, input);
+                degrade, timeline, keyCallback, keyExp, parameters, templateId, input, inputType, outputType);
         this.switches = switches;
         this.setNext(switches.stream().map(Switch::getNext).collect(Collectors.joining(",")));
     }

rill-flow-dag/olympicene-core/src/main/java/com/weibo/rill/flow/olympicene/core/model/task/TaskCategory.java

@@ -29,9 +29,6 @@ public enum TaskCategory {
     // 流程控制Task，执行分支语句
     CHOICE("choice", 1),
 
-    // 流程控制Task，执行分支语句
-    SWITCH("switch", 1),
-
     // 流程控制Task，执行循环语句
     FOREACH("foreach", 1),
 
@@ -43,6 +40,9 @@ public enum TaskCategory {
 
     // return task
     RETURN("return", 2),
+
+    // 执行分支语句
+    SWITCH("switch", 2),
     ;
 
     private final String value;
@@ -66,6 +66,7 @@ public static TaskCategory getEnumByValue(String category) {
             case "suspense" -> SUSPENSE;
             case "pass" -> PASS;
             case "return" -> RETURN;
+            case "switch" -> SWITCH;
             default -> null;
         };
     }