Send packets continuously

main.go

@@ -50,7 +50,7 @@ func main() {
 	}
 
 	if config.Respondd.Enable {
-		multiCollector = respond.NewMultiCollector(func(coll *respond.Collector, res *respond.Response) {
+		multiCollector = respond.NewMultiCollector(collectInterval, func(coll *respond.Collector, res *respond.Response) {
 
 			switch coll.CollectType {
 			case "neighbours":
@@ -75,7 +75,6 @@ func main() {
 				log.Println("unknown CollectType:", coll.CollectType)
 			}
 		})
-		go multiCollector.ListenAndSend(collectInterval)
 	}
 
 	//TODO bad

respond/collector.go

@@ -12,10 +12,16 @@ type Collector struct {
 	connection  *net.UDPConn   // UDP socket
 	queue       chan *Response // received responses
 	parse       func(coll *Collector, res *Response)
+
+	// Ticker and stopper
+	ticker *time.Ticker
+	stop   chan interface{}
 }
 
+type ParseFunc func(coll *Collector, res *Response)
+
 //NewCollector creates a Collector struct
-func NewCollector(CollectType string, parseFunc func(coll *Collector, res *Response)) *Collector {
+func NewCollector(CollectType string, interval time.Duration, parseFunc ParseFunc) *Collector {
 	// Parse address
 	addr, err := net.ResolveUDPAddr("udp", "[::]:0")
 	if err != nil {
@@ -34,18 +40,23 @@ func NewCollector(CollectType string, parseFunc func(coll *Collector, res *Respo
 		connection:  conn,
 		queue:       make(chan *Response, 400),
 		parse:       parseFunc,
+		ticker:      time.NewTicker(interval),
+		stop:        make(chan interface{}, 1),
 	}
 
 	go collector.receiver()
 	go collector.parser()
-
+	go collector.sender()
-	collector.sendOnce()
 
 	return collector
 }
 
-//Close Collector
+// Close Collector
 func (coll *Collector) Close() {
+	// stop ticker
+	coll.ticker.Stop()
+	coll.stop <- nil
+
 	coll.connection.Close()
 	close(coll.queue)
 }
@@ -66,12 +77,15 @@ func (coll *Collector) sendPacket(address string) {
 	}
 }
 
-func (coll *Collector) sender(collectInterval time.Duration) {
+// send packets continously
-	c := time.Tick(collectInterval)
+func (coll *Collector) sender() {
-
+	for {
-	for range c {
+		select {
-		// TODO break condition
+		case <-coll.stop:
-		coll.sendOnce()
+			return
+		case <-coll.ticker.C:
+			coll.sendOnce()
+		}
 	}
 }
 
@@ -98,6 +112,5 @@ func (coll *Collector) receiver() {
 			Address: *src,
 			Raw:     raw,
 		}
-		log.Println("received", coll.CollectType, "from", src)
 	}
 }

respond/multi_collector.go

@@ -8,23 +8,16 @@ type MultiCollector struct {
 }
 
 //NewMultiCollector create a list of collectors
-func NewMultiCollector(parseFunc func(coll *Collector, res *Response)) *MultiCollector {
+func NewMultiCollector(interval time.Duration, parseFunc ParseFunc) *MultiCollector {
 	return &MultiCollector{
 		collectors: []*Collector{
-			NewCollector("statistics", parseFunc),
+			NewCollector("statistics", interval, parseFunc),
-			NewCollector("nodeinfo", parseFunc),
+			NewCollector("nodeinfo", interval, parseFunc),
-			NewCollector("neighbours", parseFunc),
+			NewCollector("neighbours", interval, parseFunc),
 		},
 	}
 }
 
-//ListenAndSend on Collection
-func (multi *MultiCollector) ListenAndSend(collectInterval time.Duration) {
-	for _, col := range multi.collectors {
-		col.sender(collectInterval)
-	}
-}
-
 //Close all Collections
 func (multi *MultiCollector) Close() {
 	for _, col := range multi.collectors {