-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathexample_collector_tcp_test.go
67 lines (56 loc) · 1.6 KB
/
example_collector_tcp_test.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
package ipfix_test
import (
"bytes"
"context"
"fmt"
"log"
"os"
"os/signal"
"syscall"
"github.com/zoomoid/go-ipfix"
)
// Collect IPFIX messages via TCP listener. The code's layout is idiomatic of the package:
// Consume messages in a goroutine from a channel and use a Decoder instance to create
// message objects to work with.
// The example code simply logs the messages to Stdout. Other use-cases might be forwarding
// these objects in a stateless format, e.g., JSON or Protobuf, to a message queue, such as Kafka
// (this is what similar Go libraries such as goflow2 and vflow) do.
func Example_collectorTCP() {
var (
BindAddr string = "[::]:4739"
)
ctx, cancel := context.WithCancel(context.TODO())
defer cancel()
c := make(chan os.Signal, 2)
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
go func() {
<-c
log.Println("Received shutdown signal, initiating shutdown...")
cancel()
<-c
os.Exit(1)
}()
tcpListener := ipfix.NewTCPListener(BindAddr)
go func() {
log.Printf("Starting TCP listener for IPFIX messages on %s", BindAddr)
tcpListener.Listen(ctx)
}()
templateCache := ipfix.NewDefaultEphemeralCache()
fieldCache := ipfix.NewEphemeralFieldCache(templateCache)
decoder := ipfix.NewDecoder(templateCache, fieldCache, ipfix.DecoderOptions{OmitRFC5610Records: false})
go func() {
for {
select {
case raw := <-tcpListener.Messages():
msg, err := decoder.Decode(ctx, bytes.NewBuffer(raw))
if err != nil {
log.Println(fmt.Errorf("failed to decode IPFIX message: %w", err))
}
log.Println(msg)
case <-ctx.Done():
return
}
}
}()
<-ctx.Done()
}