forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
/
sync.go
84 lines (70 loc) · 1.62 KB
/
sync.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
package publisher
import (
"sync"
"github.com/elastic/beats/filebeat/input"
"github.com/elastic/beats/libbeat/logp"
"github.com/elastic/beats/libbeat/publisher"
)
type syncLogPublisher struct {
pub publisher.Publisher
client publisher.Client
in chan []*input.Event
out SuccessLogger
done chan struct{}
wg sync.WaitGroup
}
func newSyncLogPublisher(
in chan []*input.Event,
out SuccessLogger,
pub publisher.Publisher,
) *syncLogPublisher {
return &syncLogPublisher{
in: in,
out: out,
pub: pub,
done: make(chan struct{}),
}
}
func (p *syncLogPublisher) Start() {
p.client = p.pub.Connect()
p.wg.Add(1)
go func() {
defer p.wg.Done()
logp.Info("Start sending events to output")
defer logp.Debug("publisher", "Shutting down sync publisher")
for {
err := p.Publish()
if err != nil {
return
}
}
}()
}
func (p *syncLogPublisher) Publish() error {
var events []*input.Event
select {
case <-p.done:
return sigPublisherStop
case events = <-p.in:
}
ok := p.client.PublishEvents(getDataEvents(events), publisher.Sync, publisher.Guaranteed)
if !ok {
// PublishEvents will only returns false, if p.client has been closed.
return sigPublisherStop
}
// TODO: move counter into logger?
logp.Debug("publish", "Events sent: %d", len(events))
eventsSent.Add(int64(len(events)))
// Tell the logger that we've successfully sent these events
ok = p.out.Published(events)
if !ok {
// stop publisher if successfully send events can not be logged anymore.
return sigPublisherStop
}
return nil
}
func (p *syncLogPublisher) Stop() {
p.client.Close()
close(p.done)
p.wg.Wait()
}