-
Notifications
You must be signed in to change notification settings - Fork 16
/
task_execute_function.go
94 lines (80 loc) · 2.38 KB
/
task_execute_function.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
// 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 model
import (
"encoding/json"
"os"
"sync"
log "github.com/sirupsen/logrus"
)
var hostname string
var once sync.Once
type ExecuteTaskFunction func(t *Task) (interface{}, error)
type ValidateWorkflowFunction func(w *Workflow) (bool, error)
func NewTaskResultFromTask(task *Task) *TaskResult {
return &TaskResult{
TaskId: task.TaskId,
WorkflowInstanceId: task.WorkflowInstanceId,
WorkerId: getHostname(),
}
}
func NewTaskResultFromTaskWithError(t *Task, err error) *TaskResult {
taskResult := NewTaskResultFromTask(t)
taskResult.ReasonForIncompletion = err.Error()
switch err.(type) {
case *NonRetryableError:
taskResult.Status = FailedWithTerminalErrorTask
default:
taskResult.Status = FailedTask
}
return taskResult
}
func NewTaskResult(taskId string, workflowInstanceId string) *TaskResult {
return &TaskResult{
TaskId: taskId,
WorkflowInstanceId: workflowInstanceId,
WorkerId: getHostname(),
}
}
func GetTaskResultFromTaskExecutionOutput(t *Task, taskExecutionOutput interface{}) (*TaskResult, error) {
taskResult, ok := taskExecutionOutput.(*TaskResult)
if !ok {
taskResult = NewTaskResultFromTask(t)
outputData, err := ConvertToMap(taskExecutionOutput)
if err != nil {
return nil, err
}
taskResult.OutputData = outputData
taskResult.Status = CompletedTask
}
return taskResult, nil
}
func ConvertToMap(input interface{}) (map[string]interface{}, error) {
if input == nil {
return nil, nil
}
data, err := json.Marshal(input)
if err != nil {
log.Debug(
"Failed to parse input",
", reason: ", err.Error(),
)
return nil, err
}
var parsedInput map[string]interface{}
json.Unmarshal(data, &parsedInput)
return parsedInput, nil
}
func getHostname() string {
once.Do(updateHostname)
return hostname
}
func updateHostname() {
hostname, _ = os.Hostname()
}