-
Notifications
You must be signed in to change notification settings - Fork 242
/
conditionalrepeater.go
109 lines (97 loc) · 2.26 KB
/
conditionalrepeater.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
108
109
package transactions
import (
"context"
"sync"
"time"
)
// TaskFunc defines the task to be run. The context is canceled when Stop is
// called to early stop scheduled task.
type TaskFunc func(ctx context.Context) (done bool)
const (
WorkNotDone = false
WorkDone = true
)
// ConditionalRepeater runs a task at regular intervals until the task returns
// true. It doesn't allow running task in parallel and can be triggered early
// by call to RunUntilDone.
type ConditionalRepeater struct {
interval time.Duration
task TaskFunc
// nil if not running
ctx context.Context
ctxMu sync.Mutex
cancel context.CancelFunc
runNowCh chan bool
runNowMu sync.Mutex
}
func NewConditionalRepeater(interval time.Duration, task TaskFunc) *ConditionalRepeater {
return &ConditionalRepeater{
interval: interval,
task: task,
runNowCh: make(chan bool, 1),
}
}
// RunUntilDone starts the task immediately and continues to run it at the defined
// interval until the task returns true. Can be called multiple times but it
// does not allow multiple concurrent executions of the task.
func (t *ConditionalRepeater) RunUntilDone() {
t.ctxMu.Lock()
defer func() {
t.runNowMu.Lock()
if len(t.runNowCh) == 0 {
t.runNowCh <- true
}
t.runNowMu.Unlock()
t.ctxMu.Unlock()
}()
if t.ctx != nil {
return
}
t.ctx, t.cancel = context.WithCancel(context.Background())
go func() {
defer func() {
t.ctxMu.Lock()
defer t.ctxMu.Unlock()
t.cancel()
t.ctx = nil
}()
ticker := time.NewTicker(t.interval)
defer ticker.Stop()
for {
select {
// Stop was called or task returned true
case <-t.ctx.Done():
return
// Scheduled execution
case <-ticker.C:
if t.task(t.ctx) {
return
}
// Start right away if requested
case <-t.runNowCh:
ticker.Reset(t.interval)
if t.task(t.ctx) {
t.runNowMu.Lock()
if len(t.runNowCh) == 0 {
t.runNowMu.Unlock()
return
}
t.runNowMu.Unlock()
}
}
}
}()
}
// Stop forcefully stops the running task by canceling its context.
func (t *ConditionalRepeater) Stop() {
t.ctxMu.Lock()
defer t.ctxMu.Unlock()
if t.ctx != nil {
t.cancel()
}
}
func (t *ConditionalRepeater) IsRunning() bool {
t.ctxMu.Lock()
defer t.ctxMu.Unlock()
return t.ctx != nil
}