-
Notifications
You must be signed in to change notification settings - Fork 0
/
workflow.go
107 lines (90 loc) · 2.6 KB
/
workflow.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
package app
import (
"time"
"github.com/serverlessworkflow/sdk-go/v2/model"
"go.temporal.io/sdk/workflow"
)
type OperationState model.OperationState
// HelloWorld workflow definition
func HelloWorld(ctx workflow.Context, dslWorkflow *model.Workflow) ([]byte, error) {
logger := workflow.GetLogger(ctx)
ao := workflow.ActivityOptions{
StartToCloseTimeout: 10 * time.Second,
}
ctx = workflow.WithActivityOptions(ctx, ao)
for _, state := range dslWorkflow.States {
switch b := state.(type) {
case *model.OperationState:
err := executeOperation(ctx, b)
if err != nil {
logger.Error("DSL Workflow failed.", "Error", err)
return nil, err
}
default:
logger.Error("Unkown state")
}
}
logger.Info("DSL Workflow completed.")
return nil, nil
}
func executeOperation(ctx workflow.Context, b *model.OperationState) error {
if b.ActionMode == "parallel" {
err := executeParallel(ctx, b.Actions)
if err != nil {
return err
}
}
if b.ActionMode == "sequential" {
err := executeSequence(ctx, b.Actions)
if err != nil {
return err
}
}
return nil
}
func executeSequence(ctx workflow.Context, actions []model.Action) error {
for _, a := range actions {
err := workflow.ExecuteActivity(ctx, a.FunctionRef.RefName, a.FunctionRef.Arguments).Get(ctx, nil)
if err != nil {
return err
}
}
return nil
}
func executeParallel(ctx workflow.Context, actions []model.Action) error {
//
// You can use the context passed in to activity as a way to cancel the activity like standard GO way.
// Cancelling a parent context will cancel all the derived contexts as well.
//
// In the parallel block, we want to execute all of them in parallel and wait for all of them.
// if one activity fails then we want to cancel all the rest of them as well.
childCtx, cancelHandler := workflow.WithCancel(ctx)
selector := workflow.NewSelector(ctx)
var activityErr error
for _, s := range actions {
f := executeAsync(s, childCtx)
selector.AddFuture(f, func(f workflow.Future) {
err := f.Get(ctx, nil)
if err != nil {
// cancel all pending activities
cancelHandler()
activityErr = err
}
})
}
for i := 0; i < len(actions); i++ {
selector.Select(ctx) // this will wait for one branch
if activityErr != nil {
return activityErr
}
}
return nil
}
func executeAsync(action model.Action, ctx workflow.Context) workflow.Future {
future, settable := workflow.NewFuture(ctx)
workflow.Go(ctx, func(ctx workflow.Context) {
err := workflow.ExecuteActivity(ctx, action.FunctionRef.RefName, action.FunctionRef.Arguments).Get(ctx, nil)
settable.Set(nil, err)
})
return future
}