New UnaryTask and ConditionalTask interfaces

Co-authored-by: Matteo Nardelli <matnar@gmail.com>

Author: grussorusso

internal/test/util.go

@@ -311,7 +311,7 @@ func IsWindows() bool {
 func CreateSequenceWorkflow(funcs ...*function.Function) (*workflow.Workflow, error) {
 	builder := workflow.NewBuilder()
 	for _, f := range funcs {
-		builder = builder.AddSimpleNode(f)
+		builder = builder.AddFunctionTask(f)
 	}
 	return builder.Build()
 }

internal/test/workflow_integration_test.go

@@ -199,10 +199,10 @@ func TestInvokeFC_DifferentFunctions(t *testing.T) {
 	u.AssertNil(t, errF2)
 
 	wflow, err := workflow.NewBuilder().
-		AddSimpleNode(fDouble).
-		AddSimpleNode(fInc).
-		AddSimpleNode(fDouble).
-		AddSimpleNode(fInc).
+		AddFunctionTask(fDouble).
+		AddFunctionTask(fInc).
+		AddFunctionTask(fDouble).
+		AddFunctionTask(fInc).
 		Build()
 	wflow.Name = workflowName
 
@@ -248,7 +248,7 @@ func TestInvokeFC_Concurrent(t *testing.T) {
 	u.AssertNil(t, err)
 	builder := workflow.NewBuilder()
 	for i := 0; i < length; i++ {
-		builder.AddSimpleNodeWithId(f, fmt.Sprintf("simple %d", i))
+		builder.AddFunctionTaskWithId(f, fmt.Sprintf("simple %d", i))
 	}
 	wflow, err := builder.Build()
 	wflow.Name = workflowName
@@ -334,7 +334,7 @@ func TestInvokeSieveChoice(t *testing.T) {
 	u.AssertNil(t, errDp)
 
 	wflow, err := workflow.NewBuilder().
-		AddSimpleNode(isPrimePy).
+		AddFunctionTask(isPrimePy).
 		AddChoiceNode(
 			workflow.NewEqParamCondition(workflow.NewParam("IsPrime"), workflow.NewValue(true)),
 			workflow.NewEqParamCondition(workflow.NewParam("IsPrime"), workflow.NewValue(false)),
@@ -465,9 +465,9 @@ func TestInvokeWorkflowPassDoNothing(t *testing.T) {
 		AddOutput("result", function.Int{}).Build())
 	u.AssertNil(t, errDp)
 	wflow, err := workflow.NewBuilder().
-		AddSimpleNode(incPy).
+		AddFunctionTask(incPy).
 		AddPassNode(""). // this should not do nothing
-		AddSimpleNode(incPy).
+		AddFunctionTask(incPy).
 		Build()
 	wflow.Name = "pass_do_nothing"
 	u.AssertNil(t, err)

internal/test/workflow_test.go

@@ -71,18 +71,20 @@ func TestSimpleWorkflow(t *testing.T) {
 	var prevNode workflow.Task = wflow.Start
 	var currentNode workflow.Task
 	for !end {
-		switch prevNode.(type) {
+		switch pn := prevNode.(type) {
 		case *workflow.StartTask:
-			nextNodeId := prevNode.GetNext()
+			nextNodeId := pn.GetNext()
 			currentNode, _ = wflow.Find(nextNodeId)
-			u.AssertEquals(t, prevNode.GetNext(), currentNode.GetId())
+			u.AssertEquals(t, pn.GetNext(), currentNode.GetId())
 		case *workflow.EndTask:
 			end = true
-		default: // currentNode = simple node
-			nextNodeId := prevNode.GetNext()
+		case workflow.UnaryTask:
+			nextNodeId := pn.GetNext()
 			currentNode, _ = wflow.Find(nextNodeId)
-			u.AssertEquals(t, prevNode.GetNext(), currentNode.GetId())
+			u.AssertEquals(t, pn.GetNext(), currentNode.GetId())
 			u.AssertTrue(t, prevNode.(*workflow.FunctionTask).Func == f.Name)
+		default:
+			t.Fatalf("unknown type %T", pn)
 		}
 		prevNode = currentNode
 	}
@@ -124,11 +126,15 @@ func TestChoiceWorkflow(t *testing.T) {
 	for _, n := range wflow.Tasks {
 		switch n.(type) {
 		case *workflow.ChoiceTask:
-			u.AssertEquals(t, len(choice.Conditions), len(choice.AlternativeNextTasks))
-			for _, s := range choice.AlternativeNextTasks {
+			u.AssertEquals(t, len(choice.Conditions), len(choice.GetAlternatives()))
+			for _, s := range choice.GetAlternatives() {
 				simple, foundS := wflow.Find(s)
+				unary := simple.(*workflow.FunctionTask)
+				if unary == nil {
+					t.Fail()
+				}
 				u.AssertTrue(t, foundS)
-				u.AssertEquals(t, simple.GetNext(), wflow.End.GetId())
+				u.AssertEquals(t, unary.GetNext(), wflow.End.GetId())
 			}
 		case *workflow.FunctionTask:
 			u.AssertTrue(t, n.(*workflow.FunctionTask).Func == f.Name)
@@ -204,7 +210,7 @@ func TestWorkflowBuilder(t *testing.T) {
 	f, err := initializeExamplePyFunction()
 	u.AssertNil(t, err)
 	wflow, err := workflow.NewBuilder().
-		AddSimpleNode(f).
+		AddFunctionTask(f).
 		AddChoiceNode(workflow.NewEqCondition(1, 4), workflow.NewDiffCondition(1, 4)).
 		NextBranch(CreateSequenceWorkflow(f)).
 		NextBranch(CreateSequenceWorkflow(f)).
@@ -241,23 +247,22 @@ func TestVisit(t *testing.T) {
 	f, err := initializeExamplePyFunction()
 	u.AssertNil(t, err)
 	complexWorkflow, err := workflow.NewBuilder().
-		AddSimpleNode(f).
+		AddFunctionTask(f).
 		AddChoiceNode(workflow.NewEqCondition(1, 4), workflow.NewDiffCondition(1, 4)).
 		NextBranch(CreateSequenceWorkflow(f)).
 		NextBranch(CreateSequenceWorkflow(f, f)).
 		EndChoiceAndBuild()
 	u.AssertNil(t, err)
 
-	startNext, _ := complexWorkflow.Find(complexWorkflow.Start.GetNext())
-
-	choice := startNext.GetNext()
-
 	visitedNodes := workflow.Visit(complexWorkflow, complexWorkflow.Start.Id, false)
 	u.AssertEquals(t, len(complexWorkflow.Tasks), len(visitedNodes))
 
 	visitedNodes = workflow.Visit(complexWorkflow, complexWorkflow.Start.Id, true)
 	u.AssertEquals(t, len(complexWorkflow.Tasks)-1, len(visitedNodes))
 
+	startNext, _ := complexWorkflow.Find(complexWorkflow.Start.GetNext())
+	choice := startNext.(workflow.UnaryTask).GetNext()
+
 	visitedNodes = workflow.Visit(complexWorkflow, choice, false)
 	u.AssertEquals(t, 5, len(visitedNodes))
 

internal/workflow/asl.go

@@ -103,7 +103,7 @@ func BuildFromTaskState(builder *Builder, t *asl.TaskState, name string) (*Build
 	if !found {
 		return nil, fmt.Errorf("non existing function in workflow: %s", t.Resource)
 	}
-	builder = builder.AddSimpleNodeWithId(f, name)
+	builder = builder.AddFunctionTaskWithId(f, name)
 	return builder, nil
 }