-
Notifications
You must be signed in to change notification settings - Fork 0
/
observer.go
100 lines (88 loc) · 1.96 KB
/
observer.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
package eventd
import (
"context"
"time"
)
// Observer ...
type Observer struct {
repo Repository
errorDuration time.Duration
fetchChan chan<- fetchRequest
fromSequence uint64
limit uint64
cacheResult []Event
responseChan chan fetchResponse
logger func(err error)
}
func newObserver(
repo Repository, fetchChan chan<- fetchRequest,
fromSequence uint64, limit uint64,
options ...ObserverOption,
) *Observer {
opts := observerOpts{
logger: func(err error) {},
// TODO Configurable
errorSleepDuration: 30 * time.Second,
}
for _, o := range options {
o(&opts)
}
return &Observer{
repo: repo,
errorDuration: opts.errorSleepDuration,
fetchChan: fetchChan,
fromSequence: fromSequence,
limit: limit,
cacheResult: make([]Event, 0, limit),
responseChan: make(chan fetchResponse, 1),
logger: opts.logger,
}
}
func (o *Observer) beforeGetEvents(ctx context.Context) {
req := fetchRequest{
from: o.fromSequence,
limit: o.limit,
result: o.cacheResult,
responseChan: o.responseChan,
}
select {
case o.fetchChan <- req:
return
case <-ctx.Done():
return
}
}
func (o *Observer) waitForEvents(ctx context.Context, result []Event) ([]Event, error) {
select {
case resp := <-o.responseChan:
if !resp.existed {
events, err := o.repo.GetEventsFrom(ctx, o.fromSequence, o.limit)
if err != nil {
o.logger(err)
return result, err
}
resp.result = events
}
o.fromSequence = resp.result[len(resp.result)-1].Sequence + 1
result = append(result, resp.result...)
return result, nil
case <-ctx.Done():
return result, nil
}
}
// GetNextEvents ...
func (o *Observer) GetNextEvents(ctx context.Context, result []Event) []Event {
for {
o.beforeGetEvents(ctx)
var err error
result, err = o.waitForEvents(ctx, result)
if ctx.Err() != nil {
return result
}
if err != nil {
sleepContext(ctx, o.errorDuration)
continue
}
return result
}
}