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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
package common
import (
"context"
"fmt"
log "github.com/sirupsen/logrus"
)
// Warning that implements `error` but safe to ignore
type Warning struct {
Message string
}
// Error the contract for error
func (w Warning) Error() string {
return w.Message
}
// Warningf create a warning
func Warningf(format string, args ...interface{}) Warning {
w := Warning{
Message: fmt.Sprintf(format, args...),
}
return w
}
// Executor define contract for the steps of a workflow
type Executor func(ctx context.Context) error
// Conditional define contract for the conditional predicate
type Conditional func(ctx context.Context) bool
// NewInfoExecutor is an executor that logs messages
func NewInfoExecutor(format string, args ...interface{}) Executor {
return func(ctx context.Context) error {
logger := Logger(ctx)
logger.Infof(format, args...)
return nil
}
}
// NewPipelineExecutor creates a new executor from a series of other executors
func NewPipelineExecutor(executors ...Executor) Executor {
if len(executors) == 0 {
return func(ctx context.Context) error {
return nil
}
}
var rtn Executor
for _, executor := range executors {
if rtn == nil {
rtn = executor
} else {
rtn = rtn.Then(executor)
}
}
return rtn
}
// NewConditionalExecutor creates a new executor based on conditions
func NewConditionalExecutor(conditional Conditional, trueExecutor Executor, falseExecutor Executor) Executor {
return func(ctx context.Context) error {
if conditional(ctx) {
if trueExecutor != nil {
return trueExecutor(ctx)
}
} else {
if falseExecutor != nil {
return falseExecutor(ctx)
}
}
return nil
}
}
// NewErrorExecutor creates a new executor that always errors out
func NewErrorExecutor(err error) Executor {
return func(ctx context.Context) error {
return err
}
}
// NewParallelExecutor creates a new executor from a parallel of other executors
func NewParallelExecutor(executors ...Executor) Executor {
return func(ctx context.Context) error {
errChan := make(chan error)
for _, executor := range executors {
e := executor
go func() {
err := e.ChannelError(errChan)(ctx)
if err != nil {
log.Fatal(err)
}
}()
}
for i := 0; i < len(executors); i++ {
select {
case <-ctx.Done():
return ctx.Err()
case err := <-errChan:
if err != nil {
return err
}
}
}
return nil
}
}
// ChannelError sends error to errChan rather than returning error
func (e Executor) ChannelError(errChan chan error) Executor {
return func(ctx context.Context) error {
errChan <- e(ctx)
return nil
}
}
// Then runs another executor if this executor succeeds
func (e Executor) Then(then Executor) Executor {
return func(ctx context.Context) error {
err := e(ctx)
if err != nil {
switch err.(type) {
case Warning:
log.Warning(err.Error())
default:
log.Debugf("%+v", err)
return err
}
}
if ctx.Err() != nil {
return ctx.Err()
}
return then(ctx)
}
}
// If only runs this executor if conditional is true
func (e Executor) If(conditional Conditional) Executor {
return func(ctx context.Context) error {
if conditional(ctx) {
return e(ctx)
}
return nil
}
}
// IfNot only runs this executor if conditional is true
func (e Executor) IfNot(conditional Conditional) Executor {
return func(ctx context.Context) error {
if !conditional(ctx) {
return e(ctx)
}
return nil
}
}
// IfBool only runs this executor if conditional is true
func (e Executor) IfBool(conditional bool) Executor {
return e.If(func(ctx context.Context) bool {
return conditional
})
}
// Finally adds an executor to run after other executor
func (e Executor) Finally(finally Executor) Executor {
return func(ctx context.Context) error {
err := e(ctx)
err2 := finally(ctx)
if err2 != nil {
return fmt.Errorf("Error occurred running finally: %v (original error: %v)", err2, err)
}
return err
}
}
// Not return an inverted conditional
func (c Conditional) Not() Conditional {
return func(ctx context.Context) bool {
return !c(ctx)
}
}
|
