forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 8
/
outlet.go
59 lines (50 loc) · 1.37 KB
/
outlet.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
package channel
import (
"github.com/elastic/beats/filebeat/util"
"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common/atomic"
)
type outlet struct {
wg eventCounter
client beat.Client
isOpen atomic.Bool
}
func newOutlet(client beat.Client, wg eventCounter) *outlet {
o := &outlet{
wg: wg,
client: client,
isOpen: atomic.MakeBool(true),
}
return o
}
func (o *outlet) Close() error {
isOpen := o.isOpen.Swap(false)
if isOpen {
return o.client.Close()
}
return nil
}
func (o *outlet) OnEvent(d *util.Data) bool {
if !o.isOpen.Load() {
return false
}
event := d.GetEvent()
if d.HasState() {
event.Private = d.GetState()
}
if o.wg != nil {
o.wg.Add(1)
}
o.client.Publish(event)
// Note: race condition on shutdown:
// The underlying beat.Client is asynchronous. Without proper ACK
// handler we can not tell if the event made it 'through' or the client
// close has been completed before sending. In either case,
// we report 'false' here, indicating the event eventually being dropped.
// Returning false here, prevents the harvester from updating the state
// to the most recently published events. Therefore, on shutdown the harvester
// might report an old/outdated state update to the registry, overwriting the
// most recently
// published offset in the registry on shutdown.
return o.isOpen.Load()
}