group/node.go at master · oatcatx/group · GitHub

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
package group

import (
	"context"
	"fmt"
	"time"
)

type baseNode interface {
	Key() any
	Dep() []any
	WeakDep() []any
}

type Node interface {
	baseNode
	Exec(ctx context.Context, shared any) error
}

type AutoNode interface {
	baseNode
	Exec(ctx context.Context, shared any) (any, error)
}

type node struct {
	idx              int
	key              any
	deps, to, weakTo []int // dependencies | to nodes | weak to nodes
	f                NodeFunc
	nodeSpec
	*Group
}

type NodeFunc = func(ctx context.Context, shared any) error
type NodeErrFunc = func(ctx context.Context, shared any, err error) error

// node level interceptor
type NodeConditionFunc = func(ctx context.Context, shared any) bool
type NodePreFunc = NodeFunc
type NodeAfterFunc = NodeErrFunc
type NodeRollbackFunc = NodeErrFunc

type nodeSpec struct {
	ff       bool // fast-fail flag
	sf       bool // silent-fail flag
	retry    int
	cond     NodeConditionFunc
	pre      NodePreFunc
	after    NodeAfterFunc
	rollback NodeRollbackFunc
	timeout  time.Duration
}

func (n *node) Key(key any) *node {
	if _, ok := n.idxMap[key]; ok {
		panic(fmt.Sprintf("duplicate node key %q", key))
	}
	n.key, n.idxMap[key] = key, n.idx
	return n
}

func (n *node) Dep(keys ...any) *node {
	for _, key := range keys {
		idx, ok := n.idxMap[key]
		if !ok {
			panic(fmt.Sprintf("missing dependency %q -> %q", n.key, key))
		}
		for _, depIdx := range n.deps {
			if depIdx == idx {
				panic(fmt.Sprintf("duplicate dependency %q -> %q", n.key, key))
			}
		}
		n.deps, n.nodes[idx].to = append(n.deps, idx), append(n.nodes[idx].to, n.idx)
	}
	return n
}

func (n *node) WeakDep(keys ...any) *node {
	for _, key := range keys {
		idx, ok := n.idxMap[key]
		if !ok {
			panic(fmt.Sprintf("missing dependency %q -> %q", n.key, key))
		}
		for _, depIdx := range n.deps {
			if depIdx == idx {
				panic(fmt.Sprintf("duplicate dependency %q -> %q", n.key, key))
			}
		}
		n.deps, n.nodes[idx].to, n.nodes[idx].weakTo = append(n.deps, idx), append(n.nodes[idx].to, n.idx), append(n.nodes[idx].weakTo, n.idx)
	}
	return n
}

func (n *node) FastFail() *node {
	n.ff = true
	return n
}

func (n *node) SilentFail() *node {
	n.sf = true
	return n
}

func (n *node) WithRetry(times int) *node {
	if times < 0 {
		panic("retry times must be non-negative")
	}
	n.retry = times
	return n
}

func (n *node) WithPreFunc(f NodePreFunc) *node {
	n.pre = f
	return n
}

func (n *node) WithAfterFunc(f NodeAfterFunc) *node {
	n.after = f
	return n
}

func (n *node) WithRollback(f NodeRollbackFunc) *node {
	n.rollback = f
	return n
}

func (n *node) WithCondition(f NodeConditionFunc) *node {
	n.cond = f
	return n
}

func (n *node) SkipIf(skip bool) *node {
	if skip {
		n.cond = func(ctx context.Context, shared any) bool { return false }
	}
	return n
}

func (n *node) WithTimeout(t time.Duration) *node {
	if t <= 0 {
		panic("timeout must be positive")
	}
	n.timeout = t
	return n
}

func (n *node) Verify(panicking bool) *node {
	n.Group.Verify(panicking)
	return n
}