lutron-control/sacn.go

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()
	}
}