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 rutina
import (
"context"
"log"
"os"
"os/signal"
"sync"
"sync/atomic"
)
//Rutina is routine manager
type Rutina struct {
ctx context.Context
Cancel func()
wg sync.WaitGroup
o sync.Once
err error
logger *log.Logger
counter *uint64
cancelByError bool
}
// New instance with builtin context
func New(mixins ...Mixin) *Rutina {
ctx, cancel := context.WithCancel(context.Background())
var counter uint64 = 0
r := &Rutina{ctx: ctx, Cancel: cancel, counter: &counter, cancelByError: false}
return r.With(mixins...)
}
func (r *Rutina) With(mixins ...Mixin) *Rutina {
nr := *r
for _, m := range mixins {
m.apply(&nr)
}
return &nr
}
// Go routine
func (r *Rutina) Go(doer func(ctx context.Context) error, opts ...Options) {
onFail := ShutdownIfFail
for _, o := range opts {
switch o {
case ShutdownIfFail:
onFail = ShutdownIfFail
case RestartIfFail:
onFail = RestartIfFail
case DoNothingIfFail:
onFail = DoNothingIfFail
}
}
onDone := DoNothingIfDone
for _, o := range opts {
switch o {
case ShutdownIfDone:
onDone = ShutdownIfDone
case RestartIfDone:
onDone = RestartIfDone
case DoNothingIfDone:
onDone = DoNothingIfDone
}
}
r.wg.Add(1)
go func() {
defer r.wg.Done()
// Check that context is not canceled yet
if r.ctx.Err() != nil {
return
}
id := atomic.AddUint64(r.counter, 1)
if r.logger != nil {
r.logger.Printf("starting #%d", id)
}
if err := doer(r.ctx); err != nil {
if r.logger != nil {
r.logger.Printf("error at #%d : %v", id, err)
}
switch onFail {
case ShutdownIfFail:
if r.logger != nil {
r.logger.Printf("stopping #%d", id)
}
// Save error only if shutdown all routines
r.o.Do(func() {
r.err = err
})
r.Cancel()
case RestartIfFail:
// TODO maybe store errors on restart?
if r.logger != nil {
r.logger.Printf("restarting #%d", id)
}
r.Go(doer, opts...)
case DoNothingIfFail:
// TODO maybe store errors on nothing to do?
if r.logger != nil {
r.logger.Printf("stopping #%d", id)
}
}
} else {
switch onDone {
case ShutdownIfDone:
if r.logger != nil {
r.logger.Printf("stopping #%d with shutdown", id)
}
r.Cancel()
case RestartIfDone:
if r.logger != nil {
r.logger.Printf("restarting #%d", id)
}
r.Go(doer, opts...)
case DoNothingIfDone:
if r.logger != nil {
r.logger.Printf("stopping #%d", id)
}
}
}
}()
}
// OS signals handler
func (r *Rutina) ListenOsSignals(signals ...os.Signal) {
if len(signals) == 0 {
signals = []os.Signal{os.Kill, os.Interrupt}
}
r.Go(func(ctx context.Context) error {
sig := make(chan os.Signal, 1)
signal.Notify(sig, signals...)
select {
case s := <-sig:
if r.logger != nil {
r.logger.Printf("stopping by OS signal (%v)", s)
}
if r.cancelByError {
r.Cancel()
}
case <-ctx.Done():
}
return nil
}, ShutdownIfDone)
}
// Wait all routines and returns first error or nil if all routines completes without errors
func (r *Rutina) Wait() error {
r.wg.Wait()
return r.err
}
|
