forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
/
acker.go
41 lines (33 loc) · 754 Bytes
/
acker.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
package beater
import (
"github.com/elastic/beats/filebeat/util"
"github.com/elastic/beats/libbeat/publisher/beat"
)
// eventAcker handles publisher pipeline ACKs and forwards
// them to the registrar.
type eventACKer struct {
out successLogger
}
type successLogger interface {
Published(events []*util.Data) bool
}
func newEventACKer(out successLogger) *eventACKer {
return &eventACKer{out: out}
}
func (a *eventACKer) ackEvents(events []beat.Event) {
data := make([]*util.Data, 0, len(events))
for _, event := range events {
p := event.Private
if p == nil {
continue
}
datum, ok := p.(*util.Data)
if !ok || !datum.HasState() {
continue
}
data = append(data, datum)
}
if len(data) > 0 {
a.out.Published(data)
}
}