Removes unnecessary PrepareOutput()

Author: grussorusso

internal/function/signature.go

@@ -66,7 +66,7 @@ type OutputDef struct {
 	Type string // the type of the output parameter
 }
 
-// CheckInput evaluates all given outputs and if there is no output that type-checks, with the given name and the current type, returns an error
+// CheckOutput evaluates all given outputs and if there is no output that type-checks, with the given name and the current type, returns an error
 func (o OutputDef) CheckOutput(inputMap map[string]interface{}) error {
 	val, exists := inputMap[o.Name]
 	if !exists {

internal/workflow/choice_task.go

@@ -88,18 +88,6 @@ func (workflow *Workflow) executeChoice(progress *Progress, input *PartialData,
 	outputData.ForTask = nextTaskId
 	outputData.Data = input.Data
 
-	nextTask, ok := workflow.Find(nextTaskId)
-	if !ok {
-		return nil, progress, false, fmt.Errorf("node not found while preparing output")
-	}
-	switch typedTask := nextTask.(type) {
-	case *SimpleNode:
-		err := typedTask.MapOutput(outputData.Data)
-		if err != nil {
-			return outputData, progress, false, fmt.Errorf("choice task %s cannot prepare the output for simple task: %v", task.String(), err)
-		}
-	}
-
 	// for task node, we skip all branch that will not be executed
 	nodesToSkip := task.GetNodesToSkip(workflow, matchedCondition)
 	errSkip := progress.SkipAll(nodesToSkip)

internal/workflow/simple_task.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/labstack/gommon/log"
 	"github.com/lithammer/shortuuid"
 	"github.com/serverledge-faas/serverledge/internal/function"
 	"github.com/serverledge-faas/serverledge/internal/node"
@@ -40,11 +39,6 @@ func (workflow *Workflow) executeSimple(progress *Progress, input *PartialData,
 		return nil, progress, false, err
 	}
 
-	errSend := task.PrepareOutput(workflow, output)
-	if errSend != nil {
-		return nil, progress, false, fmt.Errorf("the node %s cannot send the output2: %v", task.String(), errSend)
-	}
-
 	nextTask := task.GetNext()[0]
 	outputData := NewPartialData(ReqId(r.Id), nextTask, task.Id, output)
 
@@ -151,65 +145,6 @@ func (s *SimpleNode) CheckInput(input map[string]interface{}) error {
 	return funct.Signature.CheckOrMatchInputs(input)
 }
 
-// PrepareOutput is used to send the output to the following function and if needed can be used to modify the SimpleNode output representation, like OutputPath
-// TODO: this logic might be integrated in input checking
-func (s *SimpleNode) PrepareOutput(workflow *Workflow, output map[string]interface{}) error {
-	funct, exists := function.GetFunction(s.Func) // we are getting the function from cache if not already downloaded
-	if !exists {
-		return fmt.Errorf("funtion %s doesn't exists", s.Func)
-	}
-	if funct.Signature == nil {
-		funct.Signature = function.SignatureInference(output)
-		log.Warnf("signature of function %s is nil. Output map: %v\n", funct.Name, output)
-	}
-	err := funct.Signature.CheckAllOutputs(output)
-	if err != nil {
-		return fmt.Errorf("error while checking outputs: %v", err)
-	}
-	// get signature of next nodes, if present and maps the output there
-
-	// we have only one output node
-	task, _ := workflow.Find(s.GetNext()[0])
-
-	switch nodeType := task.(type) {
-	case *SimpleNode:
-		return nodeType.MapOutput(output) // needed to convert type of data from one node to the next so that its signature type-checks
-	}
-
-	return nil
-}
-
-// MapOutput transforms the output for the next simpleNode, to make it compatible with its Signature
-func (s *SimpleNode) MapOutput(output map[string]interface{}) error {
-	funct, exists := function.GetFunction(s.Func)
-	if !exists {
-		return fmt.Errorf("function %s doesn't exist", s.Func)
-	}
-	sign := funct.Signature
-
-	for _, def := range sign.GetInputs() {
-		_, present := output[def.Name]
-		if present {
-			continue
-		} else if len(sign.GetInputs()) > 1 {
-			return fmt.Errorf("key %s does not exist in output data: %v", def.Name, output)
-		}
-
-		// find an entry in the output map that successfully checks the type of the InputDefinition
-		key, ok := def.FindEntryThatTypeChecks(output)
-		if ok {
-			val := output[key]
-			delete(output, key)
-			output[def.Name] = val
-		} else {
-			// otherwise if no one of the entry typechecks we are doomed
-			return fmt.Errorf("no output entry input-checks with the next function")
-		}
-	}
-
-	return nil
-}
-
 func (s *SimpleNode) GetNext() []TaskId {
 	// we only have one output
 	return []TaskId{s.OutputTo}

internal/workflow/workflow.go

@@ -213,13 +213,6 @@ func (workflow *Workflow) executeParallel(progress *Progress, input *PartialData
 					return
 				}
 
-				errSend := simpleTask.PrepareOutput(workflow, output)
-				if errSend != nil {
-					errorChannels[i] <- fmt.Errorf("the node %s cannot send the output2: %v", currTask.String(), errSend)
-					outputChannels[i] <- nil
-					return
-				}
-
 				errorChannels[i] <- nil
 				outputChannels[i] <- output
 			} else {
@@ -462,26 +455,6 @@ func (workflow *Workflow) Invoke(r *Request) (ExecutionReport, error) {
 		}
 	}
 
-	// TODO: we may need to handle offloading decisions here
-	//// saving partialData and progress on etcd - implementing workflow offloading policies
-	//err = savePartialDataToEtcd(pd)
-	//if err != nil {
-	//	return ExecutionReport{}, err
-	//}
-	//err = saveProgressToEtcd(progress)
-	//if err != nil {
-	//	return ExecutionReport{}, err
-	//}
-
-	//// deleting progresses and partial datas from cache and etcd
-	//err = DeleteProgress(requestId, cache.Persist)
-	//if err != nil {
-	//	return ExecutionReport{}, err
-	//}
-	//_, errDel := DeleteAllPartialData(requestId, cache.Persist)
-	//if errDel != nil {
-	//	return ExecutionReport{}, errDel
-	//}
 	r.ExecReport.Result = pd.Data
 
 	return r.ExecReport, nil