Fix #238 - Add support to fork task

Signed-off-by: Ricardo Zanini <ricardozanini@gmail.com>

Author: ricardozanini

impl/ctx/context.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/serverlessworkflow/sdk-go/v3/impl/utils"
 	"sync"
 	"time"
 
@@ -71,6 +72,7 @@ type WorkflowContext interface {
 	SetLocalExprVars(vars map[string]interface{})
 	AddLocalExprVars(vars map[string]interface{})
 	RemoveLocalExprVars(keys ...string)
+	Clone() WorkflowContext
 }
 
 // workflowContext holds the necessary data for the workflow execution within the instance.
@@ -118,6 +120,38 @@ func GetWorkflowContext(ctx context.Context) (WorkflowContext, error) {
 	return wfCtx, nil
 }
 
+func (ctx *workflowContext) Clone() WorkflowContext {
+	ctx.mu.Lock()
+	defer ctx.mu.Unlock()
+
+	newInput := utils.DeepCloneValue(ctx.input)
+	newOutput := utils.DeepCloneValue(ctx.output)
+
+	// deep clone each of the maps
+	newContextMap := utils.DeepClone(ctx.context)
+	newWorkflowDesc := utils.DeepClone(ctx.workflowDescriptor)
+	newTaskDesc := utils.DeepClone(ctx.taskDescriptor)
+	newLocalExprVars := utils.DeepClone(ctx.localExprVars)
+
+	newStatusPhase := append([]StatusPhaseLog(nil), ctx.StatusPhase...)
+	
+	newTasksStatusPhase := make(map[string][]StatusPhaseLog, len(ctx.TasksStatusPhase))
+	for taskName, logs := range ctx.TasksStatusPhase {
+		newTasksStatusPhase[taskName] = append([]StatusPhaseLog(nil), logs...)
+	}
+
+	return &workflowContext{
+		input:              newInput,
+		output:             newOutput,
+		context:            newContextMap,
+		workflowDescriptor: newWorkflowDesc,
+		taskDescriptor:     newTaskDesc,
+		localExprVars:      newLocalExprVars,
+		StatusPhase:        newStatusPhase,
+		TasksStatusPhase:   newTasksStatusPhase,
+	}
+}
+
 func (ctx *workflowContext) SetStartedAt(t time.Time) {
 	ctx.mu.Lock()
 	defer ctx.mu.Unlock()

impl/expr/expr.go

@@ -132,3 +132,28 @@ func mergeContextInVars(nodeCtx context.Context, variables map[string]interface{
 
 	return nil
 }
+
+func TraverseAndEvaluateObj(runtimeExpr *model.ObjectOrRuntimeExpr, input interface{}, taskName string, wfCtx context.Context) (output interface{}, err error) {
+	if runtimeExpr == nil {
+		return input, nil
+	}
+	output, err = TraverseAndEvaluate(runtimeExpr.AsStringOrMap(), input, wfCtx)
+	if err != nil {
+		return nil, model.NewErrExpression(err, taskName)
+	}
+	return output, nil
+}
+
+func TraverseAndEvaluateBool(runtimeExpr string, input interface{}, wfCtx context.Context) (bool, error) {
+	if len(runtimeExpr) == 0 {
+		return false, nil
+	}
+	output, err := TraverseAndEvaluate(runtimeExpr, input, wfCtx)
+	if err != nil {
+		return false, nil
+	}
+	if result, ok := output.(bool); ok {
+		return result, nil
+	}
+	return false, nil
+}

impl/runner.go

@@ -17,6 +17,8 @@ package impl
 import (
 	"context"
 	"fmt"
+	"github.com/serverlessworkflow/sdk-go/v3/impl/expr"
+	"github.com/serverlessworkflow/sdk-go/v3/impl/utils"
 	"time"
 
 	"github.com/serverlessworkflow/sdk-go/v3/impl/ctx"
@@ -53,6 +55,18 @@ type workflowRunnerImpl struct {
 	RunnerCtx ctx.WorkflowContext
 }
 
+func (wr *workflowRunnerImpl) CloneWithContext(newCtx context.Context) TaskSupport {
+	clonedWfCtx := wr.RunnerCtx.Clone()
+
+	ctxWithWf := ctx.WithWorkflowContext(newCtx, clonedWfCtx)
+
+	return &workflowRunnerImpl{
+		Workflow:  wr.Workflow,
+		Context:   ctxWithWf,
+		RunnerCtx: clonedWfCtx,
+	}
+}
+
 func (wr *workflowRunnerImpl) RemoveLocalExprVars(keys ...string) {
 	wr.RunnerCtx.RemoveLocalExprVars(keys...)
 }
@@ -175,13 +189,13 @@ func (wr *workflowRunnerImpl) wrapWorkflowError(err error) error {
 func (wr *workflowRunnerImpl) processInput(input interface{}) (output interface{}, err error) {
 	if wr.Workflow.Input != nil {
 		if wr.Workflow.Input.Schema != nil {
-			if err = validateSchema(input, wr.Workflow.Input.Schema, "/"); err != nil {
+			if err = utils.ValidateSchema(input, wr.Workflow.Input.Schema, "/"); err != nil {
 				return nil, err
 			}
 		}
 
 		if wr.Workflow.Input.From != nil {
-			output, err = traverseAndEvaluate(wr.Workflow.Input.From, input, "/", wr.Context)
+			output, err = expr.TraverseAndEvaluateObj(wr.Workflow.Input.From, input, "/", wr.Context)
 			if err != nil {
 				return nil, err
 			}
@@ -196,13 +210,13 @@ func (wr *workflowRunnerImpl) processOutput(output interface{}) (interface{}, er
 	if wr.Workflow.Output != nil {
 		if wr.Workflow.Output.As != nil {
 			var err error
-			output, err = traverseAndEvaluate(wr.Workflow.Output.As, output, "/", wr.Context)
+			output, err = expr.TraverseAndEvaluateObj(wr.Workflow.Output.As, output, "/", wr.Context)
 			if err != nil {
 				return nil, err
 			}
 		}
 		if wr.Workflow.Output.Schema != nil {
-			if err := validateSchema(output, wr.Workflow.Output.Schema, "/"); err != nil {
+			if err := utils.ValidateSchema(output, wr.Workflow.Output.Schema, "/"); err != nil {
 				return nil, err
 			}
 		}

impl/runner_test.go

@@ -456,3 +456,14 @@ func TestSwitchTaskRunner_DefaultCase(t *testing.T) {
 		runWorkflowTest(t, workflowPath, input, expectedOutput)
 	})
 }
+
+func TestForkSimple_NoCompete(t *testing.T) {
+	t.Run("Create a color array", func(t *testing.T) {
+		workflowPath := "./testdata/fork_simple.yaml"
+		input := map[string]interface{}{}
+		expectedOutput := map[string]interface{}{
+			"colors": []interface{}{"red", "blue"},
+		}
+		runWorkflowTest(t, workflowPath, input, expectedOutput)
+	})
+}

impl/task_runner.go

@@ -53,4 +53,7 @@ type TaskSupport interface {
 	AddLocalExprVars(vars map[string]interface{})
 	// RemoveLocalExprVars removes local variables added in AddLocalExprVars or SetLocalExprVars
 	RemoveLocalExprVars(keys ...string)
+	// CloneWithContext returns a full clone of this TaskSupport, but using
+	// the provided context.Context (so deadlines/cancellations propagate).
+	CloneWithContext(ctx context.Context) TaskSupport
 }

impl/task_runner_do.go

@@ -16,6 +16,8 @@ package impl
 
 import (
 	"fmt"
+	"github.com/serverlessworkflow/sdk-go/v3/impl/expr"
+	"github.com/serverlessworkflow/sdk-go/v3/impl/utils"
 	"time"
 
 	"github.com/serverlessworkflow/sdk-go/v3/impl/ctx"
@@ -35,6 +37,8 @@ func NewTaskRunner(taskName string, task model.Task, workflowDef *model.Workflow
 		return NewForTaskRunner(taskName, t)
 	case *model.CallHTTP:
 		return NewCallHttpRunner(taskName, t)
+	case *model.ForkTask:
+		return NewForkTaskRunner(taskName, t, workflowDef)
 	default:
 		return nil, fmt.Errorf("unsupported task type '%T' for task '%s'", t, taskName)
 	}
@@ -117,7 +121,7 @@ func (d *DoTaskRunner) runTasks(input interface{}, taskSupport TaskSupport) (out
 		}
 
 		taskSupport.SetTaskStatus(currentTask.Key, ctx.CompletedStatus)
-		input = deepCloneValue(output)
+		input = utils.DeepCloneValue(output)
 		idx, currentTask = d.TaskList.Next(idx)
 	}
 
@@ -126,7 +130,7 @@ func (d *DoTaskRunner) runTasks(input interface{}, taskSupport TaskSupport) (out
 
 func (d *DoTaskRunner) shouldRunTask(input interface{}, taskSupport TaskSupport, task *model.TaskItem) (bool, error) {
 	if task.GetBase().If != nil {
-		output, err := traverseAndEvaluateBool(task.GetBase().If.String(), input, taskSupport.GetContext())
+		output, err := expr.TraverseAndEvaluateBool(task.GetBase().If.String(), input, taskSupport.GetContext())
 		if err != nil {
 			return false, model.NewErrExpression(err, task.Key)
 		}
@@ -143,7 +147,7 @@ func (d *DoTaskRunner) evaluateSwitchTask(input interface{}, taskSupport TaskSup
 				defaultThen = switchCase.Then
 				continue
 			}
-			result, err := traverseAndEvaluateBool(model.NormalizeExpr(switchCase.When.String()), input, taskSupport.GetContext())
+			result, err := expr.TraverseAndEvaluateBool(model.NormalizeExpr(switchCase.When.String()), input, taskSupport.GetContext())
 			if err != nil {
 				return nil, model.NewErrExpression(err, taskKey)
 			}
@@ -199,11 +203,11 @@ func (d *DoTaskRunner) processTaskInput(task *model.TaskBase, taskInput interfac
 		return taskInput, nil
 	}
 
-	if err = validateSchema(taskInput, task.Input.Schema, taskName); err != nil {
+	if err = utils.ValidateSchema(taskInput, task.Input.Schema, taskName); err != nil {
 		return nil, err
 	}
 
-	if output, err = traverseAndEvaluate(task.Input.From, taskInput, taskName, taskSupport.GetContext()); err != nil {
+	if output, err = expr.TraverseAndEvaluateObj(task.Input.From, taskInput, taskName, taskSupport.GetContext()); err != nil {
 		return nil, err
 	}
 
@@ -216,11 +220,11 @@ func (d *DoTaskRunner) processTaskOutput(task *model.TaskBase, taskOutput interf
 		return taskOutput, nil
 	}
 
-	if output, err = traverseAndEvaluate(task.Output.As, taskOutput, taskName, taskSupport.GetContext()); err != nil {
+	if output, err = expr.TraverseAndEvaluateObj(task.Output.As, taskOutput, taskName, taskSupport.GetContext()); err != nil {
 		return nil, err
 	}
 
-	if err = validateSchema(output, task.Output.Schema, taskName); err != nil {
+	if err = utils.ValidateSchema(output, task.Output.Schema, taskName); err != nil {
 		return nil, err
 	}
 
@@ -232,12 +236,12 @@ func (d *DoTaskRunner) processTaskExport(task *model.TaskBase, taskOutput interf
 		return nil
 	}
 
-	output, err := traverseAndEvaluate(task.Export.As, taskOutput, taskName, taskSupport.GetContext())
+	output, err := expr.TraverseAndEvaluateObj(task.Export.As, taskOutput, taskName, taskSupport.GetContext())
 	if err != nil {
 		return err
 	}
 
-	if err = validateSchema(output, task.Export.Schema, taskName); err != nil {
+	if err = utils.ValidateSchema(output, task.Export.Schema, taskName); err != nil {
 		return nil
 	}
 

impl/task_runner_for.go

@@ -73,7 +73,7 @@ func (f *ForTaskRunner) Run(input interface{}, taskSupport TaskSupport) (interfa
 				return nil, err
 			}
 			if f.Task.While != "" {
-				whileIsTrue, err := traverseAndEvaluateBool(f.Task.While, forOutput, taskSupport.GetContext())
+				whileIsTrue, err := expr.TraverseAndEvaluateBool(f.Task.While, forOutput, taskSupport.GetContext())
 				if err != nil {
 					return nil, err
 				}

impl/task_runner_fork.go

@@ -0,0 +1,105 @@
+package impl
+
+import (
+	"context"
+	"fmt"
+	"github.com/serverlessworkflow/sdk-go/v3/model"
+	"sync"
+)
+
+func NewForkTaskRunner(taskName string, task *model.ForkTask, workflowDef *model.Workflow) (*ForkTaskRunner, error) {
+	if task == nil || task.Fork.Branches == nil {
+		return nil, model.NewErrValidation(fmt.Errorf("invalid Fork task %s", taskName), taskName)
+	}
+
+	var runners []TaskRunner
+	for _, branchItem := range *task.Fork.Branches {
+		r, err := NewTaskRunner(branchItem.Key, branchItem.Task, workflowDef)
+		if err != nil {
+			return nil, err
+		}
+		runners = append(runners, r)
+	}
+
+	return &ForkTaskRunner{
+		Task:          task,
+		TaskName:      taskName,
+		BranchRunners: runners,
+	}, nil
+}
+
+type ForkTaskRunner struct {
+	Task          *model.ForkTask
+	TaskName      string
+	BranchRunners []TaskRunner
+}
+
+func (f ForkTaskRunner) GetTaskName() string {
+	return f.TaskName
+}
+
+func (f ForkTaskRunner) Run(input interface{}, parentSupport TaskSupport) (interface{}, error) {
+	cancelCtx, cancel := context.WithCancel(parentSupport.GetContext())
+	defer cancel()
+
+	n := len(f.BranchRunners)
+	results := make([]interface{}, n)
+	errs := make(chan error, n)
+	done := make(chan struct{})
+	resultCh := make(chan interface{}, 1)
+
+	var (
+		wg   sync.WaitGroup
+		once sync.Once // <-- declare a Once
+	)
+
+	for i, runner := range f.BranchRunners {
+		wg.Add(1)
+		go func(i int, runner TaskRunner) {
+			defer wg.Done()
+			// **Isolate context** for each branch!
+			branchSupport := parentSupport.CloneWithContext(cancelCtx)
+
+			select {
+			case <-cancelCtx.Done():
+				return
+			default:
+			}
+
+			out, err := runner.Run(input, branchSupport)
+			if err != nil {
+				errs <- err
+				return
+			}
+			results[i] = out
+
+			if f.Task.Fork.Compete {
+				select {
+				case resultCh <- out:
+					once.Do(func() {
+						cancel()    // **signal cancellation** to all other branches
+						close(done) // signal we have a winner
+					})
+				default:
+				}
+			}
+		}(i, runner)
+	}
+
+	if f.Task.Fork.Compete {
+		select {
+		case <-done:
+			return <-resultCh, nil
+		case err := <-errs:
+			return nil, err
+		}
+	}
+
+	wg.Wait()
+	select {
+	case err := <-errs:
+		return nil, err
+	default:
+	}
+	return results, nil
+}