Refactoring run method

Signed-off-by: fjtirado <[email protected]>

Author: fjtirado

impl/task_runner_call_http.go

@@ -1,3 +1,17 @@
+// Copyright 2025 The Serverless Workflow Specification Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package impl
 
 import (

impl/task_runner_do.go

@@ -32,7 +32,7 @@ func NewTaskRunner(taskName string, task model.Task, workflowDef *model.Workflow
 	case *model.DoTask:
 		return NewDoTaskRunner(t.Do, workflowDef)
 	case *model.ForTask:
-		return NewForTaskRunner(taskName, t)
+		return NewForTaskRunner(taskName, t, workflowDef)
 	case *model.CallHTTP:
 		return NewCallHttpRunner(taskName, t)
 	default:
@@ -172,7 +172,7 @@ func (d *DoTaskRunner) runTask(input interface{}, taskSupport TaskSupport, runne
 	taskSupport.SetTaskName(taskName)
 
 	if task.Input != nil {
-		if input, err = d.processTaskInput(task, input, taskSupport); err != nil {
+		if input, err = d.processTaskInput(task, input, taskSupport, taskName); err != nil {
 			return nil, err
 		}
 	}
@@ -184,62 +184,62 @@ func (d *DoTaskRunner) runTask(input interface{}, taskSupport TaskSupport, runne
 
 	taskSupport.SetTaskRawOutput(output)
 
-	if output, err = d.processTaskOutput(task, output, taskSupport); err != nil {
+	if output, err = d.processTaskOutput(task, output, taskSupport, taskName); err != nil {
 		return nil, err
 	}
 
-	if err = d.processTaskExport(task, output, taskSupport); err != nil {
+	if err = d.processTaskExport(task, output, taskSupport, taskName); err != nil {
 		return nil, err
 	}
 
 	return output, nil
 }
 
 // processTaskInput processes task input validation and transformation.
-func (d *DoTaskRunner) processTaskInput(task *model.TaskBase, taskInput interface{}, taskSupport TaskSupport) (output interface{}, err error) {
+func (d *DoTaskRunner) processTaskInput(task *model.TaskBase, taskInput interface{}, taskSupport TaskSupport, taskName string) (output interface{}, err error) {
 	if task.Input == nil {
 		return taskInput, nil
 	}
 
-	if err = validateSchema(taskInput, task.Input.Schema, d.GetTaskName()); err != nil {
+	if err = validateSchema(taskInput, task.Input.Schema, taskName); err != nil {
 		return nil, err
 	}
 
-	if output, err = traverseAndEvaluate(task.Input.From, taskInput, d.GetTaskName(), taskSupport.GetContext()); err != nil {
+	if output, err = traverseAndEvaluate(task.Input.From, taskInput, taskName, taskSupport.GetContext()); err != nil {
 		return nil, err
 	}
 
 	return output, nil
 }
 
 // processTaskOutput processes task output validation and transformation.
-func (d *DoTaskRunner) processTaskOutput(task *model.TaskBase, taskOutput interface{}, taskSupport TaskSupport) (output interface{}, err error) {
+func (d *DoTaskRunner) processTaskOutput(task *model.TaskBase, taskOutput interface{}, taskSupport TaskSupport, taskName string) (output interface{}, err error) {
 	if task.Output == nil {
 		return taskOutput, nil
 	}
 
-	if output, err = traverseAndEvaluate(task.Output.As, taskOutput, d.GetTaskName(), taskSupport.GetContext()); err != nil {
+	if output, err = traverseAndEvaluate(task.Output.As, taskOutput, taskName, taskSupport.GetContext()); err != nil {
 		return nil, err
 	}
 
-	if err = validateSchema(output, task.Output.Schema, d.GetTaskName()); err != nil {
+	if err = validateSchema(output, task.Output.Schema, taskName); err != nil {
 		return nil, err
 	}
 
 	return output, nil
 }
 
-func (d *DoTaskRunner) processTaskExport(task *model.TaskBase, taskOutput interface{}, taskSupport TaskSupport) (err error) {
+func (d *DoTaskRunner) processTaskExport(task *model.TaskBase, taskOutput interface{}, taskSupport TaskSupport, taskName string) (err error) {
 	if task.Export == nil {
 		return nil
 	}
 
-	output, err := traverseAndEvaluate(task.Export.As, taskOutput, d.GetTaskName(), taskSupport.GetContext())
+	output, err := traverseAndEvaluate(task.Export.As, taskOutput, taskName, taskSupport.GetContext())
 	if err != nil {
 		return err
 	}
 
-	if err = validateSchema(output, task.Export.Schema, d.GetTaskName()); err != nil {
+	if err = validateSchema(output, task.Export.Schema, taskName); err != nil {
 		return nil
 	}
 

impl/task_runner_for.go

@@ -28,12 +28,12 @@ const (
 	forTaskDefaultAt   = "$index"
 )
 
-func NewForTaskRunner(taskName string, task *model.ForTask) (*ForTaskRunner, error) {
+func NewForTaskRunner(taskName string, task *model.ForTask, workflowDef *model.Workflow) (*ForTaskRunner, error) {
 	if task == nil || task.Do == nil {
 		return nil, model.NewErrValidation(fmt.Errorf("invalid For task %s", taskName), taskName)
 	}
 
-	doRunner, err := NewDoTaskRunner(task.Do)
+	doRunner, err := NewDoTaskRunner(task.Do, workflowDef)
 	if err != nil {
 		return nil, err
 	}