-
-
Notifications
You must be signed in to change notification settings - Fork 32
/
custom.go
109 lines (90 loc) · 2.08 KB
/
custom.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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
package main
import (
"context"
"fmt"
"cuelang.org/go/cue"
"cuelang.org/go/cue/cuecontext"
"cuelang.org/go/tools/flow"
)
var CTX cue.Context
// Our tasks specified as Cue
var input = `
tasks: {
a: {
foo: 1
hello: string
}
b: {
foo: 2
}
c: {
foo: a.foo * 3
goo: b.foo * 3
}
}
`
func main() {
var err error
fmt.Println("Custom Flow Task")
// create context
ctx := cuecontext.New()
// Setup the flow Config
cfg := &flow.Config{Root: cue.ParsePath("tasks")}
// compile our input
value := ctx.CompileString(input, cue.Filename("input.cue"))
if value.Err() != nil {
fmt.Println("Error:", value.Err())
return
}
// create the workflow whiich will build the task graph
workflow := flow.New(cfg, value, TaskFactory)
// run our custom workflow
err = workflow.Run(context.Background())
if err != nil {
fmt.Println("Error:", err)
return
}
}
// This function implements the Runner interface.
// It parses Cue values, you will see all of them recursively
func TaskFactory(val cue.Value) (flow.Runner, error) {
// You can see the recursive values with this
fmt.Println("TF: ", val)
// Check that we have something that looks like a task
foo := val.Lookup("foo")
if !foo.Exists() {
return nil, nil
}
num, err := foo.Int64()
if err != nil {
return nil, err
}
// Create and return a flow.Runner
ct := &CustomTask{
Val: int(num),
}
return ct, nil
}
// Our custom task with some extra data
// While we only have one task here, you can have as many as you like
// It's up to the TaskFunc (TaskFactory above) to create the tasks
type CustomTask struct {
Val int
}
// Tasks must implement a Run func, this is where we execute our task
func (C *CustomTask) Run(t *flow.Task, pErr error) error {
// not sure this is OK, but the value which was used for this task
val := t.Value()
fmt.Println("CustomTask:", C.Val, val)
// Do some work
next := map[string]interface{}{
"bar": C.Val + 1,
}
hello := val.LookupPath(cue.ParsePath("foo"))
if hello.Exists() {
next["hello"] = "world"
}
// Use fill to "return" a result to the workflow engine
t.Fill(next)
return nil
}