adding to git.i0t.app
This commit is contained in:
Levi Woodard
316
internal/daemon/daemon.go
Normal file
316
internal/daemon/daemon.go
Normal file
@@ -0,0 +1,316 @@
|
||||
// Package daemon is the orchestrator: it reads the config, asks the compositor
|
||||
// for the output geometry, launches one supervised mpv per occupied grid slot,
|
||||
// tiles them, and exposes a control socket so the layout can be switched live.
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/lwoodard/rtsp-streamer/internal/compositor"
|
||||
"github.com/lwoodard/rtsp-streamer/internal/config"
|
||||
"github.com/lwoodard/rtsp-streamer/internal/ipc"
|
||||
"github.com/lwoodard/rtsp-streamer/internal/player"
|
||||
)
|
||||
|
||||
// Daemon owns the running video wall.
|
||||
type Daemon struct {
|
||||
cfgPath string
|
||||
log *slog.Logger
|
||||
comp *compositor.Client
|
||||
runDir string
|
||||
|
||||
mu sync.Mutex
|
||||
cfg *config.Config
|
||||
players []*player.Player
|
||||
layout string
|
||||
cancelLo context.CancelFunc // cancels the current layout's supervisors
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// New constructs a daemon bound to a config path.
|
||||
func New(cfgPath string, log *slog.Logger) (*Daemon, error) {
|
||||
comp, err := compositor.New()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
runDir := ipc.RunDir()
|
||||
if err := os.MkdirAll(runDir, 0o700); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Daemon{cfgPath: cfgPath, log: log, comp: comp, runDir: runDir}, nil
|
||||
}
|
||||
|
||||
// Run starts the wall and blocks until ctx is cancelled.
|
||||
func (d *Daemon) Run(ctx context.Context) error {
|
||||
if err := d.comp.PrepareForMPV(ctx); err != nil {
|
||||
d.log.Warn("could not preinstall mpv window rules", "err", err)
|
||||
}
|
||||
if err := d.reload(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
go d.serveControl(ctx)
|
||||
go d.healthLoop(ctx)
|
||||
|
||||
<-ctx.Done()
|
||||
d.log.Info("shutting down")
|
||||
d.stopLayout()
|
||||
return nil
|
||||
}
|
||||
|
||||
// resolution returns the render size, from config or the live output.
|
||||
func (d *Daemon) resolution(ctx context.Context, cfg *config.Config) (int, int) {
|
||||
if cfg.Display.Width > 0 && cfg.Display.Height > 0 {
|
||||
return cfg.Display.Width, cfg.Display.Height
|
||||
}
|
||||
if out, err := d.comp.PrimaryOutput(ctx); err == nil && out.CurrentMode.Width > 0 {
|
||||
return out.CurrentMode.Width, out.CurrentMode.Height
|
||||
}
|
||||
d.log.Warn("falling back to 1920x1080; set display.width/height to override")
|
||||
return 1920, 1080
|
||||
}
|
||||
|
||||
// reload re-reads config from disk and applies the active layout.
|
||||
func (d *Daemon) reload(ctx context.Context) error {
|
||||
cfg, err := config.Load(d.cfgPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.mu.Lock()
|
||||
d.cfg = cfg
|
||||
d.mu.Unlock()
|
||||
name := cfg.ActiveLayout
|
||||
if name == "" {
|
||||
d.log.Warn("no active_layout set; nothing to display")
|
||||
d.stopLayout()
|
||||
return nil
|
||||
}
|
||||
return d.applyLayout(ctx, name)
|
||||
}
|
||||
|
||||
// applyLayout tears down the current wall and builds the named layout.
|
||||
func (d *Daemon) applyLayout(ctx context.Context, name string) error {
|
||||
d.mu.Lock()
|
||||
cfg := d.cfg
|
||||
d.mu.Unlock()
|
||||
|
||||
layout := cfg.LayoutByName(name)
|
||||
if layout == nil {
|
||||
return fmt.Errorf("layout %q not found", name)
|
||||
}
|
||||
cols, rows, err := layout.Dimensions()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w, h := d.resolution(ctx, cfg)
|
||||
tiles := layout.EffectiveTiles()
|
||||
|
||||
d.stopLayout()
|
||||
|
||||
loCtx, cancel := context.WithCancel(ctx)
|
||||
var players []*player.Player
|
||||
|
||||
for slot, tile := range tiles {
|
||||
if tile.Camera == "" {
|
||||
continue
|
||||
}
|
||||
cam := cfg.CameraByName(tile.Camera)
|
||||
if cam == nil || cam.Disabled || cam.RTSP == "" {
|
||||
d.log.Warn("skipping tile: camera unavailable", "slot", slot, "camera", tile.Camera)
|
||||
continue
|
||||
}
|
||||
cs, rs := tile.Span()
|
||||
rect := compositor.TileRect(w, h, cols, rows, tile.Col, tile.Row, cs, rs)
|
||||
p := player.New(slot, cam.Name, cam.RTSP, cfg.Player, d.runDir, d.log)
|
||||
players = append(players, p)
|
||||
|
||||
d.wg.Add(1)
|
||||
go func() { defer d.wg.Done(); p.Supervise(loCtx) }()
|
||||
|
||||
// Position the window once it maps. Done per-tile so a slow camera
|
||||
// doesn't block the others.
|
||||
d.wg.Add(1)
|
||||
go func(p *player.Player, rect compositor.Rect) {
|
||||
defer d.wg.Done()
|
||||
d.placeWhenReady(loCtx, p, rect)
|
||||
}(p, rect)
|
||||
}
|
||||
|
||||
d.mu.Lock()
|
||||
d.players = players
|
||||
d.layout = name
|
||||
d.cancelLo = cancel
|
||||
d.cfg.ActiveLayout = name
|
||||
d.mu.Unlock()
|
||||
d.log.Info("layout applied", "layout", name, "tiles", len(players), "grid", layout.Grid, "res", fmt.Sprintf("%dx%d", w, h))
|
||||
return nil
|
||||
}
|
||||
|
||||
// placeWhenReady waits for the mpv window to map then tiles it, retrying while
|
||||
// the layout is active (Supervise may relaunch mpv with a new pid).
|
||||
func (d *Daemon) placeWhenReady(ctx context.Context, p *player.Player, rect compositor.Rect) {
|
||||
var lastPID int
|
||||
ticker := time.NewTicker(500 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
pid := p.PID()
|
||||
if pid != 0 && pid != lastPID {
|
||||
if err := d.comp.WaitForWindow(ctx, pid, 15*time.Second); err == nil {
|
||||
if err := d.comp.Place(ctx, pid, rect); err != nil {
|
||||
d.log.Warn("place failed", "slot", p.Slot, "err", err)
|
||||
} else {
|
||||
lastPID = pid
|
||||
d.log.Debug("window placed", "slot", p.Slot, "pid", pid, "rect", rect)
|
||||
}
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// stopLayout cancels supervisors and kills current mpv processes.
|
||||
func (d *Daemon) stopLayout() {
|
||||
d.mu.Lock()
|
||||
cancel := d.cancelLo
|
||||
players := d.players
|
||||
d.cancelLo = nil
|
||||
d.players = nil
|
||||
d.mu.Unlock()
|
||||
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
for _, p := range players {
|
||||
p.Stop()
|
||||
}
|
||||
d.wg.Wait()
|
||||
}
|
||||
|
||||
// healthLoop periodically nudges stalled streams. Supervise already restarts
|
||||
// exited mpv; this catches the "process alive but frozen" case.
|
||||
func (d *Daemon) healthLoop(ctx context.Context) {
|
||||
ticker := time.NewTicker(20 * time.Second)
|
||||
defer ticker.Stop()
|
||||
stalls := map[int]int{}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
d.mu.Lock()
|
||||
players := append([]*player.Player(nil), d.players...)
|
||||
d.mu.Unlock()
|
||||
for _, p := range players {
|
||||
if !p.Running() {
|
||||
continue
|
||||
}
|
||||
if p.Healthy() {
|
||||
stalls[p.Slot] = 0
|
||||
continue
|
||||
}
|
||||
stalls[p.Slot]++
|
||||
if stalls[p.Slot] >= 2 {
|
||||
d.log.Warn("stream stalled, forcing restart", "slot", p.Slot, "camera", p.Name)
|
||||
p.Stop() // Supervise relaunches
|
||||
stalls[p.Slot] = 0
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// serveControl accepts control-socket connections for status/reload/set-layout.
|
||||
func (d *Daemon) serveControl(ctx context.Context) {
|
||||
path := ipc.SocketPath()
|
||||
_ = os.Remove(path)
|
||||
ln, err := net.Listen("unix", path)
|
||||
if err != nil {
|
||||
d.log.Error("control socket listen failed", "err", err)
|
||||
return
|
||||
}
|
||||
go func() { <-ctx.Done(); ln.Close(); os.Remove(path) }()
|
||||
d.log.Info("control socket ready", "path", path)
|
||||
for {
|
||||
conn, err := ln.Accept()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
go d.handleControl(ctx, conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Daemon) handleControl(ctx context.Context, conn net.Conn) {
|
||||
defer conn.Close()
|
||||
_ = conn.SetDeadline(time.Now().Add(30 * time.Second))
|
||||
var req ipc.Request
|
||||
if err := json.NewDecoder(conn).Decode(&req); err != nil {
|
||||
return
|
||||
}
|
||||
resp := d.dispatch(ctx, req)
|
||||
_ = json.NewEncoder(conn).Encode(resp)
|
||||
}
|
||||
|
||||
func (d *Daemon) dispatch(ctx context.Context, req ipc.Request) ipc.Response {
|
||||
switch req.Cmd {
|
||||
case "status":
|
||||
return d.status()
|
||||
case "reload":
|
||||
if err := d.reload(ctx); err != nil {
|
||||
return ipc.Response{OK: false, Error: err.Error()}
|
||||
}
|
||||
return d.status()
|
||||
case "set-layout":
|
||||
if err := d.applyLayout(ctx, req.Name); err != nil {
|
||||
return ipc.Response{OK: false, Error: err.Error()}
|
||||
}
|
||||
// Persist the choice so a restart keeps it.
|
||||
if err := d.persistActiveLayout(req.Name); err != nil {
|
||||
d.log.Warn("could not persist active layout", "err", err)
|
||||
}
|
||||
return d.status()
|
||||
default:
|
||||
return ipc.Response{OK: false, Error: fmt.Sprintf("unknown command %q", req.Cmd)}
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Daemon) persistActiveLayout(name string) error {
|
||||
cfg, err := config.Load(d.cfgPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cfg.ActiveLayout = name
|
||||
return config.Save(d.cfgPath, cfg)
|
||||
}
|
||||
|
||||
func (d *Daemon) status() ipc.Response {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
resp := ipc.Response{OK: true, ActiveLayout: d.layout}
|
||||
for _, p := range d.players {
|
||||
resp.Slots = append(resp.Slots, ipc.SlotStatus{
|
||||
Slot: p.Slot,
|
||||
Camera: p.Name,
|
||||
PID: p.PID(),
|
||||
Running: p.Running(),
|
||||
Healthy: p.Healthy(),
|
||||
})
|
||||
}
|
||||
return resp
|
||||
}
|
||||
Reference in New Issue
Block a user