forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
/
testing.go
71 lines (58 loc) · 1.41 KB
/
testing.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
package testing
// ChanClient implements Client interface, forwarding published events to some
import (
"github.com/elastic/beats/libbeat/beat"
)
type TestPublisher struct {
client beat.Client
}
// given channel only.
type ChanClient struct {
done chan struct{}
Channel chan beat.Event
}
func PublisherWithClient(client beat.Client) beat.Pipeline {
return &TestPublisher{client}
}
func (pub *TestPublisher) Connect() (beat.Client, error) {
return pub.client, nil
}
func (pub *TestPublisher) ConnectWith(_ beat.ClientConfig) (beat.Client, error) {
return pub.client, nil
}
func (pub *TestPublisher) SetACKHandler(_ beat.PipelineACKHandler) error {
panic("Not supported")
}
func NewChanClient(bufSize int) *ChanClient {
return NewChanClientWith(make(chan beat.Event, bufSize))
}
func NewChanClientWith(ch chan beat.Event) *ChanClient {
if ch == nil {
ch = make(chan beat.Event, 1)
}
c := &ChanClient{
done: make(chan struct{}),
Channel: ch,
}
return c
}
func (c *ChanClient) Close() error {
close(c.done)
return nil
}
// PublishEvent will publish the event on the channel. Options will be ignored.
// Always returns true.
func (c *ChanClient) Publish(event beat.Event) {
select {
case <-c.done:
case c.Channel <- event:
}
}
func (c *ChanClient) PublishAll(event []beat.Event) {
for _, e := range event {
c.Publish(e)
}
}
func (c *ChanClient) ReceiveEvent() beat.Event {
return <-c.Channel
}