Add sACN universe discovery listener
This commit is contained in:
Ian Gulliver
@@ -15,6 +15,7 @@ func main() {
|
||||
noIGMP := flag.Bool("no-igmp", false, "disable IGMP querier")
|
||||
noMDNS := flag.Bool("no-mdns", false, "disable mDNS discovery")
|
||||
noArtNet := flag.Bool("no-artnet", false, "disable Art-Net discovery")
|
||||
noSACN := flag.Bool("no-sacn", false, "disable sACN discovery")
|
||||
noDante := flag.Bool("no-dante", false, "disable Dante discovery")
|
||||
noBMD := flag.Bool("no-bmd", false, "disable Blackmagic discovery")
|
||||
noShure := flag.Bool("no-shure", false, "disable Shure discovery")
|
||||
@@ -27,6 +28,7 @@ func main() {
|
||||
debugIGMP := flag.Bool("debug-igmp", false, "debug IGMP querier")
|
||||
debugMDNS := flag.Bool("debug-mdns", false, "debug mDNS discovery")
|
||||
debugArtNet := flag.Bool("debug-artnet", false, "debug Art-Net discovery")
|
||||
debugSACN := flag.Bool("debug-sacn", false, "debug sACN discovery")
|
||||
debugDante := flag.Bool("debug-dante", false, "debug Dante discovery")
|
||||
debugBMD := flag.Bool("debug-bmd", false, "debug Blackmagic discovery")
|
||||
debugShure := flag.Bool("debug-shure", false, "debug Shure discovery")
|
||||
@@ -43,6 +45,7 @@ func main() {
|
||||
t.DisableIGMP = *noIGMP
|
||||
t.DisableMDNS = *noMDNS
|
||||
t.DisableArtNet = *noArtNet
|
||||
t.DisableSACN = *noSACN
|
||||
t.DisableDante = *noDante
|
||||
t.DisableBMD = *noBMD
|
||||
t.DisableShure = *noShure
|
||||
@@ -55,6 +58,7 @@ func main() {
|
||||
t.DebugIGMP = *debugIGMP
|
||||
t.DebugMDNS = *debugMDNS
|
||||
t.DebugArtNet = *debugArtNet
|
||||
t.DebugSACN = *debugSACN
|
||||
t.DebugDante = *debugDante
|
||||
t.DebugBMD = *debugBMD
|
||||
t.DebugShure = *debugShure
|
||||
|
||||
2
http.go
2
http.go
@@ -31,6 +31,7 @@ type StatusResponse struct {
|
||||
MulticastGroups []*MulticastGroupMembers `json:"multicast_groups"`
|
||||
ArtNetNodes []*ArtNetNode `json:"artnet_nodes"`
|
||||
SACNNodes []*SACNNode `json:"sacn_nodes"`
|
||||
SACNSources []*SACNSource `json:"sacn_sources"`
|
||||
DanteFlows []*DanteFlow `json:"dante_flows"`
|
||||
PortErrors []*PortError `json:"port_errors"`
|
||||
UnreachableNodes []string `json:"unreachable_nodes"`
|
||||
@@ -141,6 +142,7 @@ func (t *Tendrils) GetStatus() *StatusResponse {
|
||||
MulticastGroups: t.getMulticastGroups(),
|
||||
ArtNetNodes: t.getArtNetNodes(),
|
||||
SACNNodes: t.getSACNNodes(),
|
||||
SACNSources: t.getSACNSources(),
|
||||
DanteFlows: t.getDanteFlows(),
|
||||
PortErrors: t.errors.GetErrors(),
|
||||
UnreachableNodes: t.errors.GetUnreachableNodes(),
|
||||
|
||||
225
sacn_discovery.go
Normal file
225
sacn_discovery.go
Normal file
@@ -0,0 +1,225 @@
|
||||
package tendrils
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"log"
|
||||
"net"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/fvbommel/sortorder"
|
||||
"golang.org/x/net/ipv4"
|
||||
)
|
||||
|
||||
const (
|
||||
sacnPort = 5568
|
||||
vectorRootE131Extended = 0x00000008
|
||||
vectorE131Discovery = 0x00000002
|
||||
vectorUniverseDiscovery = 0x00000001
|
||||
)
|
||||
|
||||
var sacnDiscoveryAddr = net.IPv4(239, 255, 250, 214)
|
||||
|
||||
var sacnPacketIdentifier = [12]byte{
|
||||
0x41, 0x53, 0x43, 0x2d, 0x45, 0x31, 0x2e, 0x31, 0x37, 0x00, 0x00, 0x00,
|
||||
}
|
||||
|
||||
type SACNSource struct {
|
||||
TypeID string `json:"typeid"`
|
||||
Node *Node `json:"node"`
|
||||
SourceName string `json:"source_name"`
|
||||
CID string `json:"cid"`
|
||||
Universes []int `json:"universes"`
|
||||
LastSeen time.Time `json:"last_seen"`
|
||||
}
|
||||
|
||||
type SACNSources struct {
|
||||
mu sync.RWMutex
|
||||
sources map[string]*SACNSource
|
||||
}
|
||||
|
||||
func NewSACNSources() *SACNSources {
|
||||
return &SACNSources{
|
||||
sources: map[string]*SACNSource{},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SACNSources) Update(cid [16]byte, sourceName string, universes []int, srcIP net.IP) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
cidStr := formatCID(cid)
|
||||
existing, exists := s.sources[cidStr]
|
||||
if exists {
|
||||
existing.SourceName = sourceName
|
||||
existing.Universes = universes
|
||||
existing.LastSeen = time.Now()
|
||||
} else {
|
||||
s.sources[cidStr] = &SACNSource{
|
||||
TypeID: newTypeID("sacnsource"),
|
||||
SourceName: sourceName,
|
||||
CID: cidStr,
|
||||
Universes: universes,
|
||||
LastSeen: time.Now(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SACNSources) SetNode(cid string, node *Node) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if source, exists := s.sources[cid]; exists {
|
||||
source.Node = node
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SACNSources) Expire() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
expireTime := time.Now().Add(-60 * time.Second)
|
||||
for cid, source := range s.sources {
|
||||
if source.LastSeen.Before(expireTime) {
|
||||
delete(s.sources, cid)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SACNSources) GetAll() []*SACNSource {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
|
||||
result := make([]*SACNSource, 0, len(s.sources))
|
||||
for _, source := range s.sources {
|
||||
result = append(result, source)
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func formatCID(cid [16]byte) string {
|
||||
return strings.ToLower(formatUUID(cid))
|
||||
}
|
||||
|
||||
func formatUUID(b [16]byte) string {
|
||||
return strings.ToUpper(
|
||||
strings.Join([]string{
|
||||
encodeHex(b[0:4]),
|
||||
encodeHex(b[4:6]),
|
||||
encodeHex(b[6:8]),
|
||||
encodeHex(b[8:10]),
|
||||
encodeHex(b[10:16]),
|
||||
}, "-"),
|
||||
)
|
||||
}
|
||||
|
||||
func encodeHex(b []byte) string {
|
||||
const hexChars = "0123456789ABCDEF"
|
||||
result := make([]byte, len(b)*2)
|
||||
for i, v := range b {
|
||||
result[i*2] = hexChars[v>>4]
|
||||
result[i*2+1] = hexChars[v&0x0f]
|
||||
}
|
||||
return string(result)
|
||||
}
|
||||
|
||||
func (t *Tendrils) startSACNDiscoveryListener(ctx context.Context, iface net.Interface) {
|
||||
c, err := net.ListenPacket("udp4", ":5568")
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] failed to listen sacn discovery: %v", err)
|
||||
return
|
||||
}
|
||||
defer c.Close()
|
||||
|
||||
p := ipv4.NewPacketConn(c)
|
||||
|
||||
if err := p.JoinGroup(&iface, &net.UDPAddr{IP: sacnDiscoveryAddr}); err != nil {
|
||||
log.Printf("[ERROR] failed to join sacn discovery multicast on %s: %v", iface.Name, err)
|
||||
return
|
||||
}
|
||||
|
||||
if t.DebugSACN {
|
||||
log.Printf("[sacn] listening for discovery on %s", iface.Name)
|
||||
}
|
||||
|
||||
buf := make([]byte, 1500)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
c.SetReadDeadline(time.Now().Add(1 * time.Second))
|
||||
n, _, err := c.ReadFrom(buf)
|
||||
if err != nil {
|
||||
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
|
||||
continue
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
t.handleSACNDiscoveryPacket(buf[:n])
|
||||
}
|
||||
}
|
||||
|
||||
func (t *Tendrils) handleSACNDiscoveryPacket(data []byte) {
|
||||
if len(data) < 120 {
|
||||
return
|
||||
}
|
||||
|
||||
if data[4] != sacnPacketIdentifier[0] || data[5] != sacnPacketIdentifier[1] ||
|
||||
data[6] != sacnPacketIdentifier[2] || data[7] != sacnPacketIdentifier[3] {
|
||||
return
|
||||
}
|
||||
|
||||
rootVector := binary.BigEndian.Uint32(data[18:22])
|
||||
if rootVector != vectorRootE131Extended {
|
||||
return
|
||||
}
|
||||
|
||||
framingVector := binary.BigEndian.Uint32(data[40:44])
|
||||
if framingVector != vectorE131Discovery {
|
||||
return
|
||||
}
|
||||
|
||||
var cid [16]byte
|
||||
copy(cid[:], data[22:38])
|
||||
|
||||
sourceName := strings.TrimRight(string(data[44:108]), "\x00")
|
||||
|
||||
discoveryVector := binary.BigEndian.Uint32(data[114:118])
|
||||
if discoveryVector != vectorUniverseDiscovery {
|
||||
return
|
||||
}
|
||||
|
||||
universeCount := (len(data) - 120) / 2
|
||||
universes := make([]int, 0, universeCount)
|
||||
for i := 0; i < universeCount; i++ {
|
||||
u := binary.BigEndian.Uint16(data[120+i*2 : 122+i*2])
|
||||
if u >= 1 && u <= 63999 {
|
||||
universes = append(universes, int(u))
|
||||
}
|
||||
}
|
||||
|
||||
if t.DebugSACN {
|
||||
log.Printf("[sacn] discovery from %q cid=%s universes=%v", sourceName, formatCID(cid), universes)
|
||||
}
|
||||
|
||||
t.sacnSources.Update(cid, sourceName, universes, nil)
|
||||
t.NotifyUpdate()
|
||||
}
|
||||
|
||||
func (t *Tendrils) getSACNSources() []*SACNSource {
|
||||
t.sacnSources.Expire()
|
||||
|
||||
sources := t.sacnSources.GetAll()
|
||||
sort.Slice(sources, func(i, j int) bool {
|
||||
return sortorder.NaturalLess(sources[i].SourceName, sources[j].SourceName)
|
||||
})
|
||||
|
||||
return sources
|
||||
}
|
||||
@@ -35,6 +35,7 @@ type Tendrils struct {
|
||||
nodes *Nodes
|
||||
artnet *ArtNetNodes
|
||||
artnetConn *net.UDPConn
|
||||
sacnSources *SACNSources
|
||||
danteFlows *DanteFlows
|
||||
errors *ErrorTracker
|
||||
ping *PingManager
|
||||
@@ -53,6 +54,7 @@ type Tendrils struct {
|
||||
DisableIGMP bool
|
||||
DisableMDNS bool
|
||||
DisableArtNet bool
|
||||
DisableSACN bool
|
||||
DisableDante bool
|
||||
DisableBMD bool
|
||||
DisableShure bool
|
||||
@@ -65,6 +67,7 @@ type Tendrils struct {
|
||||
DebugIGMP bool
|
||||
DebugMDNS bool
|
||||
DebugArtNet bool
|
||||
DebugSACN bool
|
||||
DebugDante bool
|
||||
DebugBMD bool
|
||||
DebugShure bool
|
||||
@@ -76,6 +79,7 @@ func New() *Tendrils {
|
||||
t := &Tendrils{
|
||||
activeInterfaces: map[string]context.CancelFunc{},
|
||||
artnet: NewArtNetNodes(),
|
||||
sacnSources: NewSACNSources(),
|
||||
danteFlows: NewDanteFlows(),
|
||||
ping: NewPingManager(),
|
||||
sseSubs: map[int]chan struct{}{},
|
||||
@@ -302,6 +306,9 @@ func (t *Tendrils) startInterface(ctx context.Context, iface net.Interface) {
|
||||
if !t.DisableArtNet {
|
||||
go t.startArtNetPoller(ctx, iface)
|
||||
}
|
||||
if !t.DisableSACN {
|
||||
go t.startSACNDiscoveryListener(ctx, iface)
|
||||
}
|
||||
if !t.DisableDante {
|
||||
go t.listenDante(ctx, iface)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user