forked from basho/riak-go-client
/
async.go
65 lines (58 loc) · 1.33 KB
/
async.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
package riak
import (
"sync"
"time"
"github.com/basho/backoff"
)
// Async object is used to pass required arguments to execute a Command asynchronously
type Async struct {
Command Command
Done chan Command
Wait *sync.WaitGroup
Error error
rb *backoff.Backoff // rb - Retry Backoff
enqueuedAt time.Time
executeAt time.Time
qb *backoff.Backoff // qb - Queue Backoff
}
func (a *Async) onExecute() {
if a.rb == nil {
a.rb = &backoff.Backoff{
Jitter: true,
}
} else {
a.rb.Reset()
}
}
func (a *Async) onRetry() {
d := a.rb.Duration()
logDebug("[Async]", "onRetry cmd: %s sleep: %v", a.Command.Name(), d)
time.Sleep(d)
}
func (a *Async) onEnqueued() {
if a.qb == nil {
a.enqueuedAt = time.Now()
a.qb = &backoff.Backoff{
Factor: 1.5,
Jitter: true,
}
}
a.executeAt = a.enqueuedAt.Add(a.qb.Duration())
}
func (a *Async) done(err error) {
if err != nil {
// TODO FUTURE evaluate debug logging
// logDebugln("[Async]", "done error:", err)
a.Error = err
}
if a.Done != nil {
// TODO FUTURE evaluate debug logging
// logDebug("[Async]", "signaling a.Done channel with '%s'", a.Command.Name())
a.Done <- a.Command
}
if a.Wait != nil {
// TODO FUTURE evaluate debug logging
// logDebug("[Async]", "signaling a.Wait WaitGroup for '%s'", a.Command.Name())
a.Wait.Done()
}
}