yanic/respond/collector.go

119 lines
2.2 KiB
Go
Raw Normal View History

2016-02-26 09:28:31 +01:00
package respond
2015-12-29 04:08:03 +01:00
import (
"log"
"net"
"time"
)
const (
2016-01-04 02:07:09 +01:00
// default multicast group used by announced
multiCastGroup string = "ff02:0:0:0:0:0:2:1001"
2016-01-04 02:07:09 +01:00
// default udp port used by announced
port string = "1001"
2016-01-04 02:07:09 +01:00
// maximum receivable size
maxDataGramSize int = 8192
2015-12-29 04:08:03 +01:00
)
2016-02-26 09:28:31 +01:00
//Response of the respond request
2016-01-04 02:07:09 +01:00
type Response struct {
Address net.UDPAddr
Raw []byte
}
2016-02-26 09:28:31 +01:00
//Collector for a specificle respond messages
2015-12-29 04:08:03 +01:00
type Collector struct {
CollectType string
2016-01-04 02:07:09 +01:00
connection *net.UDPConn // UDP socket
queue chan *Response // received responses
parse func(coll *Collector, res *Response)
2015-12-29 04:08:03 +01:00
}
//NewCollector creates a Collector struct
func NewCollector(CollectType string, parseFunc func(coll *Collector, res *Response)) *Collector {
2015-12-29 04:08:03 +01:00
// Parse address
addr, err := net.ResolveUDPAddr("udp", "[::]:0")
2015-12-29 04:08:03 +01:00
if err != nil {
log.Panic(err)
}
// Open socket
conn, err := net.ListenUDP("udp", addr)
if err != nil {
log.Panic(err)
}
conn.SetReadBuffer(maxDataGramSize)
2015-12-29 04:08:03 +01:00
collector := &Collector{
CollectType: CollectType,
2015-12-29 14:05:47 +01:00
connection: conn,
2016-02-19 14:25:11 +01:00
queue: make(chan *Response, 400),
parse: parseFunc,
2015-12-29 04:08:03 +01:00
}
go collector.receiver()
go collector.parser()
2016-02-19 17:14:14 +01:00
collector.sendOnce()
2015-12-29 04:08:03 +01:00
return collector
}
//Close Collector
2015-12-29 04:08:03 +01:00
func (coll *Collector) Close() {
coll.connection.Close()
close(coll.queue)
}
2015-12-29 14:05:47 +01:00
func (coll *Collector) sendOnce() {
coll.sendPacket(net.JoinHostPort(multiCastGroup, port))
2016-02-19 17:14:14 +01:00
log.Println("send request")
2015-12-29 14:05:47 +01:00
}
func (coll *Collector) sendPacket(address string) {
2015-12-29 04:08:03 +01:00
addr, err := net.ResolveUDPAddr("udp", address)
2016-02-19 11:13:30 +01:00
if err != nil {
log.Panic(err)
}
2015-12-29 14:05:47 +01:00
coll.connection.WriteToUDP([]byte(coll.CollectType), addr)
2015-12-29 04:08:03 +01:00
}
func (coll *Collector) sender(collectInterval time.Duration) {
2015-12-29 14:05:47 +01:00
c := time.Tick(collectInterval)
for range c {
2016-01-04 02:07:09 +01:00
// TODO break condition
2015-12-29 14:05:47 +01:00
coll.sendOnce()
2015-12-29 04:08:03 +01:00
}
}
func (coll *Collector) parser() {
2016-01-04 02:07:09 +01:00
for obj := range coll.queue {
coll.parse(coll, obj)
2015-12-29 04:08:03 +01:00
}
}
func (coll *Collector) receiver() {
buf := make([]byte, maxDataGramSize)
2015-12-29 04:08:03 +01:00
for {
2016-01-04 02:07:09 +01:00
n, src, err := coll.connection.ReadFromUDP(buf)
2015-12-29 04:08:03 +01:00
if err != nil {
log.Println("ReadFromUDP failed:", err)
2015-12-29 14:05:47 +01:00
return
2015-12-29 04:08:03 +01:00
}
2016-01-04 02:07:09 +01:00
raw := make([]byte, n)
copy(raw, buf)
coll.queue <- &Response{
Address: *src,
Raw: raw,
}
log.Println("received", coll.CollectType, "from", src)
2015-12-29 04:08:03 +01:00
}
}