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
|
package rutina
import (
"context"
"log"
"os"
"os/signal"
"sync"
"sync/atomic"
"syscall"
)
//Rutina is routine manager
type Rutina struct {
ctx context.Context // State of application (started/stopped)
Cancel func() // Cancel func that stops all routines
wg sync.WaitGroup // WaitGroup that wait all routines to complete
onceErr sync.Once // Flag that prevents overwrite first error that shutdowns all routines
onceWait sync.Once // Flag that prevents wait already waited rutina
err error // First error that shutdowns all routines
logger *log.Logger // Optional logger
counter *uint64 // Optional counter that names routines with increment ids for debug purposes at logger
errCh chan error // Optional channel for errors when RestartIfError and DoNothingIfError
lifecycleListener LifecycleListener // Optional listener for events
autoListenSignals []os.Signal // Optional listening os signals, default disabled
}
// New instance with builtin context
func New(mixins ...Mixin) *Rutina {
ctx, cancel := context.WithCancel(context.Background())
var counter uint64
r := &Rutina{ctx: ctx, Cancel: cancel, counter: &counter, errCh: nil}
return r.With(mixins...)
}
// With applies mixins
func (r *Rutina) With(mixins ...Mixin) *Rutina {
for _, m := range mixins {
m.apply(r)
}
if r.autoListenSignals != nil {
r.ListenOsSignals(r.autoListenSignals...)
}
return r
}
// Go routine
func (r *Rutina) Go(doer func(ctx context.Context) error, opts ...Options) {
// Check that context is not canceled yet
if r.ctx.Err() != nil {
return
}
onFail := ShutdownIfError
for _, o := range opts {
switch o {
case ShutdownIfError:
onFail = ShutdownIfError
case RestartIfError:
onFail = RestartIfError
case DoNothingIfError:
onFail = DoNothingIfError
}
}
onDone := ShutdownIfDone
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()
id := atomic.AddUint64(r.counter, 1)
r.lifecycleEvent(EventRoutineStart, int(id))
if err := doer(r.ctx); err != nil {
r.lifecycleEvent(EventRoutineError, int(id))
r.lifecycleEvent(EventRoutineStop, int(id))
// errors history
if r.errCh != nil {
r.errCh <- err
}
// region routine failed
switch onFail {
case ShutdownIfError:
// Save error only if shutdown all routines
r.onceErr.Do(func() {
r.err = err
})
r.Cancel()
case RestartIfError:
r.Go(doer, opts...)
}
// endregion
} else {
r.lifecycleEvent(EventRoutineComplete, int(id))
r.lifecycleEvent(EventRoutineStop, int(id))
// region routine successfully done
switch onDone {
case ShutdownIfDone:
r.Cancel()
case RestartIfDone:
r.Go(doer, opts...)
}
// endregion
}
}()
}
// Errors returns chan for all errors, event if DoNothingIfError or RestartIfError set.
// By default it nil. Use MixinErrChan to turn it on
func (r *Rutina) Errors() <-chan error {
return r.errCh
}
// ListenOsSignals is simple OS signals handler. By default listen syscall.SIGINT and syscall.SIGTERM
func (r *Rutina) ListenOsSignals(signals ...os.Signal) {
if len(signals) == 0 {
signals = []os.Signal{syscall.SIGINT, syscall.SIGTERM}
}
go func() {
sig := make(chan os.Signal, 1)
signal.Notify(sig, signals...)
r.log("starting OS signals listener")
select {
case s := <-sig:
r.log("stopping by OS signal (%v)", s)
r.Cancel()
case <-r.ctx.Done():
}
}()
}
// Wait all routines and returns first error or nil if all routines completes without errors
func (r *Rutina) Wait() error {
r.onceWait.Do(func() {
r.wg.Wait()
r.lifecycleEvent(EventAppStop, 0)
if r.err == nil {
r.lifecycleEvent(EventAppComplete, 0)
} else {
r.lifecycleEvent(EventAppError, 0)
}
if r.errCh != nil {
close(r.errCh)
}
})
return r.err
}
func (r *Rutina) lifecycleEvent(ev Event, rid int) {
r.log("Event = %s Routine ID = %d", ev.String(), rid)
if r.lifecycleListener != nil {
r.lifecycleListener(ev, rid)
}
}
// Log if can
func (r *Rutina) log(format string, args ...interface{}) {
if r.logger != nil {
r.logger.Printf(format, args...)
}
}
|
