package main import ( "context" "net" "sync" "time" "github.com/jsimonetti/go-artnet" "github.com/jsimonetti/go-artnet/packet" "github.com/jsimonetti/go-artnet/packet/code" log "github.com/sirupsen/logrus" ) // artnetWatchdogMax bounds how often the timeout watchdog wakes; the actual // interval is the smaller of this and the configured timeout so loss is detected // promptly without busy-polling. const artnetWatchdogMax = time.Second // ArtNetSource receives Art-Net DMX and maps a universe's channels to its // device's zones and scenes. type ArtNetSource struct { cfg *SourceConfig dev *Device disp *dmxDispatcher node *artnet.Node // portAddress is the 15-bit Art-Net port address (net/subnet/universe) this // source listens for. portAddress uint16 mu sync.Mutex lastRx time.Time // Time of the most recent matching packet. released bool // True once loss has been signalled, until packets resume. closeOnce sync.Once // Guards node shutdown; go-artnet's Stop panics if called twice. } // newArtNetSource constructs an Art-Net control source. func newArtNetSource(cfg *SourceConfig, dev *Device, disp *dmxDispatcher) *ArtNetSource { port := uint16(cfg.ArtNet.Net&0x7F)<<8 | uint16(cfg.ArtNet.SubNet&0x0F)<<4 | uint16(cfg.ArtNet.Universe&0x0F) return &ArtNetSource{cfg: cfg, dev: dev, disp: disp, portAddress: port} } // Name returns the source's configured name. func (s *ArtNetSource) Name() string { return s.cfg.Name } // Start opens the Art-Net node and registers the DMX callback. func (s *ArtNetSource) Start(ctx context.Context) error { ip := net.ParseIP(s.cfg.ArtNet.Bind) if ip == nil { ip = net.IPv4zero } // Drive the node's logging through our configured logger rather than its // default debug-to-stdout logger. nodeLog := artnet.NewLogger(log.NewEntry(log.StandardLogger())) node := artnet.NewNode(serviceName, code.StNode, ip, nodeLog) s.node = node node.RegisterCallback(code.OpDMX, func(p packet.ArtNetPacket) { dmx, ok := p.(*packet.ArtDMXPacket) if !ok { return } // Match the packet's port address against the configured one. addr := uint16(dmx.Net&0x7F)<<8 | uint16(dmx.SubUni) if addr != s.portAddress { return } // Mark the stream live so the watchdog holds off on releasing while frames // keep arriving, even when a sender slows retransmission of a static look. s.markRx() s.disp.dispatch(dmx.Data[:]) }) if err := node.Start(); err != nil { return err } log.Infof("[%s] Listening for Art-Net on net %d subnet %d universe %d", s.cfg.Name, s.cfg.ArtNet.Net, s.cfg.ArtNet.SubNet, s.cfg.ArtNet.Universe) // Detect source loss and stop the node when the application shuts down. go s.watchdog(ctx) go func() { <-ctx.Done() s.stop() }() return nil } // markRx records a freshly received frame, clearing any prior released state so // the watchdog resumes guarding the now-live stream. func (s *ArtNetSource) markRx() { s.mu.Lock() s.lastRx = time.Now() s.released = false s.mu.Unlock() } // watchdog releases the source when the universe falls silent. Unlike sACN, the // Art-Net library reports every received frame rather than only changes, so the // stream stays live as long as the sender retransmits; this guards the case where // it stops entirely, mirroring sACN's timeout: the owned zones black out and the // arbitration lock is allowed to lapse so a lower-priority source can resume. func (s *ArtNetSource) watchdog(ctx context.Context) { timeout := time.Duration(s.cfg.ArtNet.TimeoutSec * float64(time.Second)) if timeout <= 0 { return } interval := min(timeout, artnetWatchdogMax) ticker := time.NewTicker(interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: s.mu.Lock() // Skip until the first frame arrives, and only release once per loss. if s.lastRx.IsZero() || s.released || time.Since(s.lastRx) < timeout { s.mu.Unlock() continue } s.released = true s.mu.Unlock() s.disp.release() log.Infof("[%s] Art-Net universe %d released; zones to 0", s.cfg.Name, s.cfg.ArtNet.Universe) } } } // Stop shuts down the Art-Net node. func (s *ArtNetSource) Stop() { s.stop() } // stop shuts the Art-Net node down exactly once. go-artnet's Stop closes an // internal channel and panics on a second call, and both context cancellation // and Stop race to call it. func (s *ArtNetSource) stop() { s.closeOnce.Do(func() { if s.node != nil { s.node.Stop() } }) }