package main import ( "context" "fmt" "net" "sync" "time" "github.com/Hundemeier/go-sacn/sacn" log "github.com/sirupsen/logrus" ) // sacnKeepAliveMax bounds how often a live but unchanging stream re-asserts its // arbitration lock; the actual interval is the smaller of this and half the // configured hold so the lock never lapses while frames keep arriving. const sacnKeepAliveMax = time.Second // SACNSource receives sACN (E1.31) DMX and maps a universe's channels to its // device's zones and scenes. type SACNSource struct { cfg *SourceConfig dev *Device disp *dmxDispatcher recv *sacn.ReceiverSocket closeOnce sync.Once mu sync.Mutex live bool // True while the universe's stream is present (set on data, cleared on timeout). } // newSACNSource constructs an sACN control source. func newSACNSource(cfg *SourceConfig, dev *Device, disp *dmxDispatcher) *SACNSource { return &SACNSource{cfg: cfg, dev: dev, disp: disp} } // Name returns the source's configured name. func (s *SACNSource) Name() string { return s.cfg.Name } // Start opens the receiver socket and joins the configured universe. func (s *SACNSource) Start(ctx context.Context) error { // Resolve the multicast interface when one is configured. var ifi *net.Interface if s.cfg.SACN.Interface != "" { got, err := net.InterfaceByName(s.cfg.SACN.Interface) if err != nil { return err } ifi = got } recv, err := sacn.NewReceiverSocket(s.cfg.SACN.Bind, ifi) if err != nil { return err } s.recv = recv universe := s.cfg.SACN.Universe recv.SetOnChangeCallback(func(old sacn.DataPacket, newPacket sacn.DataPacket) { if newPacket.Universe() != universe { return } // The receiver only reports changed frames; mark the stream live so the // keepalive loop holds the lock while a static look keeps streaming. s.setLive(true) s.disp.dispatch(newPacket.Data()) }) recv.SetTimeoutCallback(func(univ uint16) { if univ != universe { return } // The console stopped streaming (released): black out the zones this // source drives on source loss, and stop refreshing the lock so a // lower-priority source can resume after hold. s.setLive(false) s.disp.release() log.Infof("[%s] sACN universe %d released; zones to 0", s.cfg.Name, univ) }) recv.Start() // JoinUniverse panics if the multicast group can't be joined (e.g. no usable // interface); recover so one source's failure doesn't crash the bridge. if err := joinUniverse(recv, universe); err != nil { recv.Close() s.recv = nil return err } log.Infof("[%s] Listening for sACN on universe %d", s.cfg.Name, universe) // Hold the arbitration lock while the stream stays live, then close the // receiver when the application shuts down. go s.keepAlive(ctx) go func() { <-ctx.Done() s.close() }() return nil } // setLive records whether the universe's stream is currently present. func (s *SACNSource) setLive(live bool) { s.mu.Lock() s.live = live s.mu.Unlock() } // keepAlive re-asserts zone activity while the stream is live so a static look // (which produces no change callbacks) doesn't lapse the source's arbitration // lock and let a lower-priority source take over. The hold window itself still // governs how long control persists after the stream actually stops. func (s *SACNSource) keepAlive(ctx context.Context) { interval := time.Duration(s.cfg.HoldSec*float64(time.Second)) / 2 if interval <= 0 || interval > sacnKeepAliveMax { interval = sacnKeepAliveMax } ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: s.mu.Lock() live := s.live s.mu.Unlock() if live { s.disp.refreshActivity() } } } } // close shuts the receiver socket down exactly once. go-sacn's Close panics on // a second call, and both context cancellation and Stop race to close it. func (s *SACNSource) close() { s.closeOnce.Do(func() { s.recv.Close() }) } // joinUniverse joins a universe, converting go-sacn's panic-on-failure into an // error the caller can act on. func joinUniverse(recv *sacn.ReceiverSocket, universe uint16) (err error) { defer func() { if r := recover(); r != nil { err = fmt.Errorf("join universe %d: %v", universe, r) } }() recv.JoinUniverse(universe) return nil } // Stop closes the receiver socket. func (s *SACNSource) Stop() { if s.recv != nil { s.close() } }