feature: Listen on all interfaces for multicast messages

Author: g3force

cmd/ssl-status-board/main.go

@@ -6,19 +6,25 @@ import (
 	"github.com/gobuffalo/packr"
 	"log"
 	"net/http"
+	"strings"
 )
 
 func main() {
 	var config = board.DefaultConfig()
 	var address = flag.String("address", config.ListenAddress, "The address on which the UI and API is served")
 	var refereeAddress = flag.String("refereeAddress", config.RefereeConnection.MulticastAddress, "The multicast address of ssl-game-controller")
+	var skipInterfaces = flag.String("skipInterfaces", "", "Comma separated list of interface names to ignore when receiving multicast packets")
 	flag.Parse()
 
 	config.ListenAddress = *address
 	config.RefereeConnection.MulticastAddress = *refereeAddress
+	if skipInterfaces != nil && *skipInterfaces != "" {
+		config.RefereeConnection.SkipInterfaces = parseSkipInterfaces(*skipInterfaces)
+	}
 
 	refereeBoard := board.NewBoard(config.RefereeConnection)
-	go refereeBoard.HandleIncomingMessages()
+	refereeBoard.MulticastReceiver.SkipInterfaces = config.RefereeConnection.SkipInterfaces
+	refereeBoard.Start()
 	http.HandleFunc(config.RefereeConnection.SubscribePath, refereeBoard.WsHandler)
 
 	setupUi(config.ListenAddress)
@@ -38,3 +44,7 @@ func setupUi(listenAddress string) {
 		log.Print("Backend-only version started. Run the UI separately or get a binary that has the UI included")
 	}
 }
+
+func parseSkipInterfaces(skipInterfaces string) []string {
+	return strings.Split(skipInterfaces, ",")
+}

pkg/board/refereeConnection.go pkg/board/board.gopkg/board/refereeConnection.go renamed to pkg/board/board.go

@@ -1,41 +1,57 @@
 package board
 
 import (
+	"github.com/RoboCup-SSL/ssl-status-board/pkg/sslnet"
 	"github.com/gorilla/websocket"
 	"log"
 	"net/http"
+	"sync"
 	"time"
 )
 
 // Board contains the state of this referee board
 type Board struct {
-	cfg         RefereeConfig
-	refereeData []byte
+	cfg               RefereeConfig
+	refereeData       []byte
+	mutex             sync.Mutex
+	MulticastReceiver *sslnet.MulticastReceiver
 }
 
 // NewBoard creates a new referee board
-func NewBoard(cfg RefereeConfig) Board {
-	return Board{cfg: cfg}
+func NewBoard(cfg RefereeConfig) (b *Board) {
+	b = new(Board)
+	b.cfg = cfg
+	b.MulticastReceiver = sslnet.NewMulticastReceiver(b.handlingMessage)
+	return
 }
 
-// HandleIncomingMessages listens for new messages and stores the latest ones
-func (b *Board) HandleIncomingMessages() {
-	HandleIncomingMessages(b.cfg.ConnectionConfig.MulticastAddress, b.handlingMessage)
+// Start listening for messages
+func (b *Board) Start() {
+	b.MulticastReceiver.Start(b.cfg.MulticastAddress)
+}
+
+// Stop listening for messages
+func (b *Board) Stop() {
+	b.MulticastReceiver.Stop()
 }
 
 func (b *Board) handlingMessage(data []byte) {
+	b.mutex.Lock()
+	defer b.mutex.Unlock()
 	b.refereeData = data
 }
 
 // SendToWebSocket sends latest data to the given websocket
 func (b *Board) SendToWebSocket(conn *websocket.Conn) {
 	for {
+		b.mutex.Lock()
 		if len(b.refereeData) > 0 {
 			if err := conn.WriteMessage(websocket.BinaryMessage, b.refereeData); err != nil {
 				log.Println("Could not write to referee websocket: ", err)
 				return
 			}
 		}
+		b.mutex.Unlock()
 
 		time.Sleep(b.cfg.SendingInterval)
 	}

pkg/board/serverConfig.go pkg/board/config.gopkg/board/serverConfig.go renamed to pkg/board/config.go

@@ -10,6 +10,7 @@ type ConnectionConfig struct {
 	SubscribePath    string        `yaml:"SubscribePath"`
 	SendingInterval  time.Duration `yaml:"SendingInterval"`
 	MulticastAddress string        `yaml:"MulticastAddress"`
+	SkipInterfaces   []string      `yaml:"skip-interfaces"`
 }
 
 // RefereeConfig contains referee specific connection parameters
@@ -39,6 +40,7 @@ func DefaultConfig() Config {
 		RefereeConnection: RefereeConfig{
 			ConnectionConfig: ConnectionConfig{
 				MulticastAddress: "224.5.23.1:10003",
+				SkipInterfaces:   []string{},
 				SendingInterval:  time.Millisecond * 100,
 				SubscribePath:    "/api/referee",
 			},

pkg/board/multicast.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()
+}