Removes FanOut/FanIn; GetNext returns single Id

Author: grussorusso

internal/test/api_test.go

@@ -210,7 +210,6 @@ func TestAsyncInvokeWorkflow(t *testing.T) {
 		t.Skip("Skipping integration test")
 	}
 	fcName := "sequence"
-	//deleteWorkflowApiTest(t, fcName, HOST, PORT)
 
 	fn, err := InitializePyFunction("inc", "handler", function.NewSignature().
 		AddInput("input", function.Int{}).
@@ -241,7 +240,6 @@ func TestAsyncInvokeWorkflow(t *testing.T) {
 	for {
 		pollResult := pollWorkflowTest(t, reqIdStruct.ReqId, HOST, PORT)
 
-		fmt.Println(pollResult)
 		var response workflow.InvocationResponse
 		errUnmarshalExecResult := json.Unmarshal([]byte(pollResult), &response)
 

internal/test/progress_test.go

@@ -36,46 +36,6 @@ func choiceProgress(t *testing.T, condition workflow.Condition) (*workflow.Progr
 	return workflow.InitProgress("abc", wflow), wflow
 }
 
-func parallelProgress(t *testing.T) (*workflow.Progress, *workflow.Workflow) {
-	py, err := initializeExamplePyFunction()
-	u.AssertNil(t, err)
-
-	wflow, err := workflow.NewBuilder().
-		AddBroadcastFanOutNode(3).
-		NextFanOutBranch(CreateSequenceWorkflow(py)).
-		NextFanOutBranch(CreateSequenceWorkflow(py, py)).
-		NextFanOutBranch(CreateSequenceWorkflow(py, py, py)).
-		AddFanInNode().
-		Build()
-	u.AssertNil(t, err)
-
-	return workflow.InitProgress("abc", wflow), wflow
-}
-
-func complexProgress(t *testing.T, condition workflow.Condition) (*workflow.Progress, *workflow.Workflow) {
-	py, err := initializeExamplePyFunction()
-	u.AssertNil(t, err)
-
-	notCondition := workflow.NewPredicate().Not(condition).Build()
-
-	wflow, err := workflow.NewBuilder().
-		AddSimpleNode(py).
-		AddChoiceNode(
-			notCondition,
-			condition,
-		).
-		NextBranch(CreateSequenceWorkflow(py)).
-		NextBranch(workflow.NewBuilder().
-			AddBroadcastFanOutNode(3).
-			ForEachParallelBranch(func() (*workflow.Workflow, error) { return CreateSequenceWorkflow(py, py) }).
-			AddFanInNode().
-			Build()).
-		EndChoiceAndBuild()
-	u.AssertNil(t, err)
-
-	return workflow.InitProgress("abc", wflow), wflow
-}
-
 func TestProgressMarshaling(t *testing.T) {
 	condition := workflow.NewPredicate().And(
 		workflow.NewEqCondition(1, 3),
@@ -84,9 +44,7 @@ func TestProgressMarshaling(t *testing.T) {
 
 	progress1, _ := simpleProgress(t)
 	progress2, _ := choiceProgress(t, condition)
-	progress3, _ := parallelProgress(t)
-	progress4, _ := complexProgress(t, condition)
-	progresses := []*workflow.Progress{progress1, progress2, progress3, progress4}
+	progresses := []*workflow.Progress{progress1, progress2}
 
 	for i, progress := range progresses {
 		marshal, errMarshal := json.Marshal(progress)
@@ -111,10 +69,8 @@ func TestProgressCache(t *testing.T) {
 
 	progress1, workflow1 := simpleProgress(t)
 	progress2, workflow2 := choiceProgress(t, condition)
-	progress3, workflow3 := parallelProgress(t)
-	progress4, workflow4 := complexProgress(t, condition)
-	progresses := []*workflow.Progress{progress1, progress2, progress3, progress4}
-	workflows := []*workflow.Workflow{workflow1, workflow2, workflow3, workflow4}
+	progresses := []*workflow.Progress{progress1, progress2}
+	workflows := []*workflow.Workflow{workflow1, workflow2}
 
 	for i := 0; i < len(workflows); i++ {
 		progress := progresses[i]
@@ -127,7 +83,7 @@ func TestProgressCache(t *testing.T) {
 		u.AssertTrueMsg(t, progress.Equals(retrievedProgress), "progresses don't match")
 
 		progress.Complete(wflow.Start.Id)
-		progress.Complete(wflow.Start.GetNext()[0])
+		progress.Complete(wflow.Start.GetNext())
 
 		err = workflow.SaveProgress(progress)
 		u.AssertNilMsg(t, err, "failed to save after update")

internal/test/util.go

@@ -324,48 +324,6 @@ func CreateChoiceWorkflow(dagger func() (*workflow.Workflow, error), condArr ...
 		EndChoiceAndBuild()
 }
 
-// CreateScatterSingleFunctionWorkflow if successful, returns a workflow with one fan out, N simple node with the same function
-// and then a fan in node that merges all the result in an array.
-func CreateScatterSingleFunctionWorkflow(fun *function.Function, fanOutDegree int) (*workflow.Workflow, error) {
-	return workflow.NewBuilder().
-		AddScatterFanOutNode(fanOutDegree).
-		ForEachParallelBranch(func() (*workflow.Workflow, error) { return CreateSequenceWorkflow(fun) }).
-		AddFanInNode().
-		Build()
-}
-
-// CreateBroadcastWorkflow if successful, returns a workflow with one fan out node, N simple nodes with different functions and a fan in node
-// The number of branches is defined by the number of given functions
-func CreateBroadcastWorkflow(dagger func() (*workflow.Workflow, error), fanOutDegree int) (*workflow.Workflow, error) {
-	return workflow.NewBuilder().
-		AddBroadcastFanOutNode(fanOutDegree).
-		ForEachParallelBranch(dagger).
-		AddFanInNode().
-		Build()
-}
-
-// CreateBroadcastMultiFunctionWorkflow if successful, returns a workflow with one fan out node, each branch chained with a different workflow that run in parallel, and a fan in node.
-// The number of branch is defined as the number of dagger functions.
-func CreateBroadcastMultiFunctionWorkflow(dagger ...func() (*workflow.Workflow, error)) (*workflow.Workflow, error) {
-	builder := workflow.NewBuilder().
-		AddBroadcastFanOutNode(len(dagger))
-	for _, dagFn := range dagger {
-		builder = builder.NextFanOutBranch(dagFn())
-	}
-	return builder.
-		AddFanInNode().
-		Build()
-}
-
-func GetSingleResult(cer *workflow.ExecutionReport) (string, error) {
-	if len(cer.Result) == 1 {
-		for _, value := range cer.Result {
-			return fmt.Sprintf("%v", value), nil
-		}
-	}
-	return "", fmt.Errorf("there is not exactly one result: there are %d result(s)", len(cer.Result))
-}
-
 func GetIntSingleResult(cer *workflow.ExecutionReport) (int, error) {
 	if len(cer.Result) == 1 {
 		for _, value := range cer.Result {

internal/test/workflow_integration_test.go

@@ -234,50 +234,6 @@ func TestInvokeFC_DifferentFunctions(t *testing.T) {
 	u.AssertNil(t, err3)
 }
 
-// TestInvokeFC_BroadcastFanOut executes a Parallel Workflow with N parallel branches
-func TestInvokeFC_BroadcastFanOut(t *testing.T) {
-	if testing.Short() {
-		t.Skip("Skipping integration test")
-	}
-
-	workflowName := "testBrFO"
-	// CREATE - we create a test function composition
-	fDouble, errF1 := InitializePyFunction("double", "handler", function.NewSignature().
-		AddInput("input", function.Int{}).
-		AddOutput("result", function.Int{}).
-		Build())
-	u.AssertNil(t, errF1)
-
-	width := 3
-	wflow, err := CreateBroadcastWorkflow(func() (*workflow.Workflow, error) { return CreateSequenceWorkflow(fDouble) }, width)
-	wflow.Name = workflowName
-	u.AssertNil(t, err)
-
-	err1 := wflow.Save()
-	u.AssertNil(t, err1)
-
-	// INVOKE - we call the function composition
-	params := make(map[string]interface{})
-	params[fDouble.Signature.GetInputs()[0].Name] = 1
-	request := workflow.NewRequest(shortuuid.New(), wflow, params)
-	request.CanDoOffloading = false
-	err2 := wflow.Invoke(request)
-	u.AssertNil(t, err2)
-
-	// check multiple result
-	output := request.ExecReport.Result
-
-	u.AssertNonNil(t, output)
-	for i := 0; i < width; i++ {
-		currOutput := output[fmt.Sprintf("%d", i)].(map[string]interface{})
-		u.AssertEquals(t, 2, currOutput["result"])
-	}
-
-	// cleaning up function composition and functions
-	//err3 := workflow.Delete()
-	//u.AssertNil(t, err3)
-}
-
 // TestInvokeFC_Concurrent executes concurrently m times a Sequential Workflow of length N, where each node executes a simple increment function.
 func TestInvokeFC_Concurrent(t *testing.T) {
 
@@ -352,50 +308,6 @@ func TestInvokeFC_Concurrent(t *testing.T) {
 	u.AssertNil(t, err3)
 }
 
-// TestInvokeFC_ScatterFanOut executes a Parallel Workflow with N parallel branches
-func TestInvokeFC_ScatterFanOut(t *testing.T) {
-	if testing.Short() {
-		t.Skip("Skipping integration test")
-	}
-	//for i := 0; i < 1; i++ {
-
-	workflowName := "test"
-	// CREATE - we create a test function composition
-	fDouble, errF1 := InitializePyFunction("double", "handler", function.NewSignature().
-		AddInput("input", function.Int{}).
-		AddOutput("result", function.Int{}).
-		Build())
-	u.AssertNil(t, errF1)
-
-	width := 3
-	wflow, err := CreateScatterSingleFunctionWorkflow(fDouble, width)
-	wflow.Name = workflowName
-	u.AssertNil(t, err)
-
-	err1 := wflow.Save()
-	u.AssertNil(t, err1)
-
-	// INVOKE - we call the function composition
-	params := make(map[string]interface{})
-	params[fDouble.Signature.GetInputs()[0].Name] = []int{1, 2, 3}
-	request := workflow.NewRequest(shortuuid.New(), wflow, params)
-	request.CanDoOffloading = false
-	err2 := wflow.Invoke(request)
-	u.AssertNil(t, err2)
-
-	// check multiple result
-	output := request.ExecReport.Result
-	u.AssertNonNil(t, output)
-	for i := 0; i < width; i++ {
-		currOutput := output[fmt.Sprintf("%d", i)].(map[string]interface{})
-		u.AssertEquals(t, (i+1)*2, cast.ToInt(currOutput["result"]))
-	}
-
-	// cleaning up function composition and functions
-	err3 := wflow.Delete()
-	u.AssertNil(t, err3)
-}
-
 func TestInvokeSieveChoice(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping integration test")