feature: Listen on all interfaces for multicast messages

Author: g3force

internal/app/rcon/server.go

@@ -15,6 +15,7 @@ import (
 	"net"
 	"os"
 	"strings"
+	"sync"
 )
 
 type Server struct {
@@ -25,6 +26,7 @@ type Server struct {
 	connectionHandler func(net.Conn)
 	listener          net.Listener
 	running           bool
+	mutex             sync.Mutex
 }
 
 type Client struct {
@@ -68,6 +70,8 @@ func (s *Server) Stop() {
 	if s.listener == nil {
 		return
 	}
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
 	s.running = false
 	if err := s.listener.Close(); err != nil {
 		log.Printf("Could not close listener: %v", err)
@@ -81,7 +85,7 @@ func (s *Server) Stop() {
 func (s *Server) listen() {
 	log.Print("Listening on ", s.address)
 
-	for s.running {
+	for s.isRunning() {
 		conn, err := s.listener.Accept()
 		if err != nil {
 			log.Print("Could not accept connection: ", err)
@@ -91,6 +95,12 @@ func (s *Server) listen() {
 	}
 }
 
+func (s *Server) isRunning() bool {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	return s.running
+}
+
 func (s *Server) createListener() (net.Listener, error) {
 	return net.Listen("tcp", s.address)
 }

internal/app/tracker/trackerReceiver.go

@@ -1,75 +1,47 @@
 package tracker
 
 import (
+	"github.com/RoboCup-SSL/ssl-game-controller/pkg/sslnet"
 	"github.com/golang/protobuf/proto"
 	"log"
-	"net"
-	"time"
+	"sync"
 )
 
-const maxDatagramSize = 8192
-
 type Receiver struct {
-	address  string
-	Callback func(*TrackerWrapperPacket)
-	conn     *net.UDPConn
+	address           string
+	Callback          func(*TrackerWrapperPacket)
+	mutex             sync.Mutex
+	MulticastReceiver *sslnet.MulticastReceiver
 }
 
 // NewReceiver creates a new receiver
 func NewReceiver(address string) (v *Receiver) {
 	v = new(Receiver)
 	v.address = address
 	v.Callback = func(*TrackerWrapperPacket) {}
+	v.MulticastReceiver = sslnet.NewMulticastReceiver(v.consumeData)
 	return
 }
 
 // Start starts the receiver
 func (v *Receiver) Start() {
-	addr, err := net.ResolveUDPAddr("udp", v.address)
-	if err != nil {
-		log.Print(err)
-		return
-	}
-	conn, err := net.ListenMulticastUDP("udp", nil, addr)
-	if err != nil {
-		log.Print(err)
-		return
-	}
-
-	if err := conn.SetReadBuffer(maxDatagramSize); err != nil {
-		log.Printf("Could not set read buffer to %v.", maxDatagramSize)
-	}
-	log.Println("Receiving tracking from", v.address)
-
-	v.conn = conn
-	go v.receive()
+	v.MulticastReceiver.Start(v.address)
 }
 
 // Stop stops the receiver
 func (v *Receiver) Stop() {
-	if err := v.conn.Close(); err != nil {
-		log.Printf("Could not close vision receiver: %v", err)
-	}
+	v.MulticastReceiver.Stop()
 }
 
-func (v *Receiver) receive() {
-	b := make([]byte, maxDatagramSize)
-	for {
-		n, err := v.conn.Read(b)
-		if err != nil {
-			log.Print("Could not read: ", err)
-			time.Sleep(1 * time.Second)
-			continue
-		}
-		if n >= maxDatagramSize {
-			log.Fatal("Buffer size too small")
-		}
-		wrapper := TrackerWrapperPacket{}
-		if err := proto.Unmarshal(b[0:n], &wrapper); err != nil {
-			log.Println("Could not unmarshal referee message")
-			continue
-		}
+func (v *Receiver) consumeData(data []byte) {
+	v.mutex.Lock()
+	defer v.mutex.Unlock()
 
-		v.Callback(&wrapper)
+	wrapper := TrackerWrapperPacket{}
+	if err := proto.Unmarshal(data, &wrapper); err != nil {
+		log.Println("Could not unmarshal referee message")
+		return
 	}
+
+	v.Callback(&wrapper)
 }

internal/app/vision/visionReceiver.go

@@ -1,21 +1,21 @@
 package vision
 
 import (
+	"github.com/RoboCup-SSL/ssl-game-controller/pkg/sslnet"
 	"github.com/RoboCup-SSL/ssl-game-controller/pkg/timer"
 	"github.com/golang/protobuf/proto"
 	"log"
-	"net"
+	"sync"
 	"time"
 )
 
-const maxDatagramSize = 8192
-
 type Receiver struct {
 	address           string
 	DetectionCallback func(*SSL_DetectionFrame)
 	GeometryCallback  func(*SSL_GeometryData)
-	conn              *net.UDPConn
 	latestTimestamp   time.Time
+	mutex             sync.Mutex
+	MulticastReceiver *sslnet.MulticastReceiver
 }
 
 // NewReceiver creates a new receiver
@@ -24,66 +24,41 @@ func NewReceiver(address string) (v *Receiver) {
 	v.address = address
 	v.DetectionCallback = func(*SSL_DetectionFrame) {}
 	v.GeometryCallback = func(data *SSL_GeometryData) {}
+	v.MulticastReceiver = sslnet.NewMulticastReceiver(v.consumeData)
 	return
 }
 
 // Start starts the receiver
 func (v *Receiver) Start() {
-	addr, err := net.ResolveUDPAddr("udp", v.address)
-	if err != nil {
-		log.Print(err)
-		return
-	}
-	conn, err := net.ListenMulticastUDP("udp", nil, addr)
-	if err != nil {
-		log.Print(err)
-		return
-	}
-
-	if err := conn.SetReadBuffer(maxDatagramSize); err != nil {
-		log.Printf("Could not set read buffer to %v.", maxDatagramSize)
-	}
-	log.Println("Receiving vision from", v.address)
-
-	v.conn = conn
-	go v.receive()
+	v.MulticastReceiver.Start(v.address)
 }
 
 // Stop stops the receiver
 func (v *Receiver) Stop() {
-	if err := v.conn.Close(); err != nil {
-		log.Printf("Could not close vision receiver: %v", err)
-	}
+	v.MulticastReceiver.Stop()
 }
 
-func (v *Receiver) receive() {
-	b := make([]byte, maxDatagramSize)
-	for {
-		n, err := v.conn.Read(b)
-		if err != nil {
-			log.Print("Could not read: ", err)
-			time.Sleep(1 * time.Second)
-			continue
-		}
-		if n >= maxDatagramSize {
-			log.Fatal("Buffer size too small")
-		}
-		wrapper := SSL_WrapperPacket{}
-		if err := proto.Unmarshal(b[0:n], &wrapper); err != nil {
-			log.Println("Could not unmarshal referee message")
-			continue
-		}
+func (v *Receiver) consumeData(data []byte) {
+	v.mutex.Lock()
+	defer v.mutex.Unlock()
+
+	wrapper := SSL_WrapperPacket{}
+	if err := proto.Unmarshal(data, &wrapper); err != nil {
+		log.Println("Could not unmarshal referee message")
+		return
+	}
 
-		if wrapper.Geometry != nil {
-			v.GeometryCallback(wrapper.Geometry)
-		}
-		if wrapper.Detection != nil {
-			v.latestTimestamp = timer.TimestampToTime(*wrapper.Detection.TCapture)
-			v.DetectionCallback(wrapper.Detection)
-		}
+	if wrapper.Geometry != nil {
+		v.GeometryCallback(wrapper.Geometry)
+	}
+	if wrapper.Detection != nil {
+		v.latestTimestamp = timer.TimestampToTime(*wrapper.Detection.TCapture)
+		v.DetectionCallback(wrapper.Detection)
 	}
 }
 
 func (v *Receiver) Time() time.Time {
+	v.mutex.Lock()
+	defer v.mutex.Unlock()
 	return v.latestTimestamp
 }

pkg/sslnet/multicast_receiver.go

@@ -0,0 +1,126 @@
+package sslnet
+
+import (
+	"log"
+	"net"
+	"sync"
+	"time"
+)
+
+const maxDatagramSize = 8192
+
+type MulticastReceiver struct {
+	activeIfis     map[string]bool
+	connections    []*net.UDPConn
+	running        bool
+	consumer       func([]byte)
+	mutex          sync.Mutex
+	SkipInterfaces []string
+}
+
+func NewMulticastReceiver(consumer func([]byte)) (r *MulticastReceiver) {
+	r = new(MulticastReceiver)
+	r.activeIfis = map[string]bool{}
+	r.consumer = consumer
+	return
+}
+
+func (r *MulticastReceiver) Start(multicastAddress string) {
+	r.running = true
+	go r.receive(multicastAddress)
+}
+
+func (r *MulticastReceiver) Stop() {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	r.running = false
+	for _, c := range r.connections {
+		if err := c.Close(); err != nil {
+			log.Println("Could not close connection: ", err)
+		}
+	}
+}
+
+func (r *MulticastReceiver) receive(multicastAddress string) {
+	for r.isRunning() {
+		ifis, _ := net.Interfaces()
+		for _, ifi := range ifis {
+			if ifi.Flags&net.FlagMulticast == 0 || // No multicast support
+				r.skipInterface(ifi.Name) {
+				continue
+			}
+			r.mutex.Lock()
+			if _, ok := r.activeIfis[ifi.Name]; !ok {
+				// interface not active, (re-)start receiving
+				go r.receiveOnInterface(multicastAddress, ifi)
+			}
+			r.mutex.Unlock()
+		}
+		time.Sleep(1 * time.Second)
+	}
+}
+
+func (r *MulticastReceiver) isRunning() bool {
+	r.mutex.Lock()
+	defer r.mutex.Unlock()
+	return r.running
+}
+
+func (r *MulticastReceiver) skipInterface(ifiName string) bool {
+	for _, skipIfi := range r.SkipInterfaces {
+		if skipIfi == ifiName {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *MulticastReceiver) receiveOnInterface(multicastAddress string, ifi net.Interface) {
+	addr, err := net.ResolveUDPAddr("udp", multicastAddress)
+	if err != nil {
+		log.Printf("Could resolve multicast address %v: %v", multicastAddress, err)
+		return
+	}
+
+	listener, err := net.ListenMulticastUDP("udp", &ifi, addr)
+	if err != nil {
+		log.Printf("Could not listen at %v: %v", multicastAddress, err)
+		return
+	}
+
+	if err := listener.SetReadBuffer(maxDatagramSize); err != nil {
+		log.Println("Could not set read buffer: ", err)
+	}
+
+	r.mutex.Lock()
+	r.connections = append(r.connections, listener)
+	r.activeIfis[ifi.Name] = true
+	r.mutex.Unlock()
+
+	log.Printf("Listening on %s (%s)", multicastAddress, ifi.Name)
+
+	data := make([]byte, maxDatagramSize)
+	for {
+		n, _, err := listener.ReadFrom(data)
+		if err != nil {
+			log.Println("ReadFromUDP failed:", err)
+			break
+		}
+
+		r.consumer(data[:n])
+	}
+
+	log.Printf("Stop listening on %s (%s)", multicastAddress, ifi.Name)
+
+	if err := listener.Close(); err != nil {
+		log.Println("Could not close listener: ", err)
+	}
+	r.mutex.Lock()
+	delete(r.activeIfis, ifi.Name)
+	for i, c := range r.connections {
+		if c == listener {
+			r.connections = append(r.connections[:i], r.connections[i+1:]...)
+		}
+	}
+	r.mutex.Unlock()
+}