forked from juju/juju
/
resumer.go
97 lines (84 loc) · 2.21 KB
/
resumer.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
// Copyright 2012, 2013 Canonical Ltd.
// Licensed under the AGPLv3, see LICENCE file for details.
package resumer
import (
"time"
"github.com/juju/errors"
"github.com/juju/juju/worker/catacomb"
"github.com/juju/loggo"
"github.com/juju/utils/clock"
)
var logger = loggo.GetLogger("juju.worker.resumer")
// Facade defines the interface for types capable of resuming
// transactions.
type Facade interface {
// ResumeTransactions resumes all pending transactions.
ResumeTransactions() error
}
// Config holds the dependencies and configuration necessary to
// drive a Resumer.
type Config struct {
Facade Facade
Clock clock.Clock
Interval time.Duration
}
// Validate returns an error if config cannot be expected to drive
// a Resumer.
func (config Config) Validate() error {
if config.Facade == nil {
return errors.NotValidf("nil Facade")
}
if config.Clock == nil {
return errors.NotValidf("nil Clock")
}
if config.Interval <= 0 {
return errors.NotValidf("non-positive Interval")
}
return nil
}
// Resumer is responsible for periodically resuming all pending
// transactions.
type Resumer struct {
catacomb catacomb.Catacomb
config Config
}
// NewResumer returns a new Resumer or an error. If the Resumer is
// not nil, the caller is responsible for stopping it via `Kill()`
// and handling any error returned from `Wait()`.
var NewResumer = func(config Config) (*Resumer, error) {
if err := config.Validate(); err != nil {
return nil, errors.Trace(err)
}
rr := &Resumer{config: config}
err := catacomb.Invoke(catacomb.Plan{
Site: &rr.catacomb,
Work: rr.loop,
})
if err != nil {
return nil, errors.Trace(err)
}
return rr, nil
}
// Kill is part of the worker.Worker interface.
func (rr *Resumer) Kill() {
rr.catacomb.Kill(nil)
}
// Wait is part of the worker.Worker interface.
func (rr *Resumer) Wait() error {
return rr.catacomb.Wait()
}
func (rr *Resumer) loop() error {
var interval time.Duration
for {
select {
case <-rr.catacomb.Dying():
return rr.catacomb.ErrDying()
case <-rr.config.Clock.After(interval):
err := rr.config.Facade.ResumeTransactions()
if err != nil {
return errors.Annotate(err, "cannot resume transactions")
}
}
interval = rr.config.Interval
}
}