feat: harden engine config, discovery and backpressure stats

- config: validate port/quality/fps/source/stream-index ranges on load
  and before engine start (flinger.Config.Validate)
- discovery: add Announce option to disable multicast (GUI checkbox,
  CLI --no-announce); absent JSON key keeps the default true
- backpressure: count dropped frames in the TCP sender, expose via
  engine Status and a live counter in the GUI status label
- docs: record M3/M4 decisions and X11 coverage via the portal backend
This commit is contained in:
2026-09-18 20:44:39 +01:00
parent cb27a0d63d
commit 2b3fd54934
8 changed files with 195 additions and 21 deletions
+2
View File
@@ -39,6 +39,7 @@ func main() {
fps = flag.Int("fps", 30, "video frames per second")
source = flag.String("source", "screen", "capture source: screen or pattern")
withAudio = flag.Bool("audio", true, "capture and stream system audio")
noAnnounce = flag.Bool("no-announce", false, "do not announce on the LAN (receiver must connect by IP)")
streamIndex = flag.Int("stream-index", 0, "monitor index to capture (screen source)")
duration = flag.Duration("duration", 0, "stream duration (0 = run until interrupted)")
)
@@ -52,6 +53,7 @@ func main() {
Source: *source,
Audio: *withAudio,
StreamIndex: *streamIndex,
Announce: !*noAnnounce,
}
eng, err := flinger.New(cfg)
+14
View File
@@ -23,11 +23,15 @@ type Config struct {
Source string `json:"source"`
Audio bool `json:"audio"`
StreamIndex int `json:"stream_index"`
// Announce is a *bool so an absent JSON key (older config files) keeps
// the default instead of silently disabling announcements.
Announce *bool `json:"announce"`
}
// Default returns the default configuration.
func Default() Config {
c := flinger.DefaultConfig()
announce := c.Announce
return Config{
Name: c.Name,
Port: c.Port,
@@ -36,11 +40,16 @@ func Default() Config {
Source: c.Source,
Audio: c.Audio,
StreamIndex: c.StreamIndex,
Announce: &announce,
}
}
// ToFlinger converts a persisted config to the engine config.
func (c Config) ToFlinger() flinger.Config {
announce := true
if c.Announce != nil {
announce = *c.Announce
}
return flinger.Config{
Name: c.Name,
Port: c.Port,
@@ -49,6 +58,7 @@ func (c Config) ToFlinger() flinger.Config {
Source: c.Source,
Audio: c.Audio,
StreamIndex: c.StreamIndex,
Announce: announce,
}
}
@@ -96,6 +106,10 @@ func Load() (Config, error) {
if c.Source == "" {
c.Source = d.Source
}
if c.Announce == nil {
announce := true
c.Announce = &announce
}
return c, nil
}
+18 -1
View File
@@ -29,6 +29,7 @@ func TestSaveLoadRoundTrip(t *testing.T) {
pathVar = filepath.Join(t.TempDir(), "config.json")
defer func() { pathVar = old }()
announce := false
want := Config{
Name: "Studio",
Port: 9898,
@@ -37,6 +38,7 @@ func TestSaveLoadRoundTrip(t *testing.T) {
Source: "pattern",
Audio: false,
StreamIndex: 1,
Announce: &announce,
}
if err := Save(want); err != nil {
t.Fatalf("Save: %v", err)
@@ -46,7 +48,10 @@ func TestSaveLoadRoundTrip(t *testing.T) {
if err != nil {
t.Fatalf("Load: %v", err)
}
if got != want {
if got.Name != want.Name || got.Port != want.Port || got.Quality != want.Quality ||
got.FPS != want.FPS || got.Source != want.Source || got.Audio != want.Audio ||
got.StreamIndex != want.StreamIndex || got.Announce == nil || want.Announce == nil ||
*got.Announce != *want.Announce {
t.Errorf("round trip mismatch:\n got %+v\nwant %+v", got, want)
}
}
@@ -74,4 +79,16 @@ func TestLoadFillsZeroValues(t *testing.T) {
if c.Quality != 80 {
t.Errorf("quality = %d, want filled default 80", c.Quality)
}
// Absent "announce" key must default to true (not false).
if c.Announce == nil || !*c.Announce {
t.Errorf("announce = %v, want default true", c.Announce)
}
}
// TestToFlingerAnnounceDefault checks the announce default survives conversion.
func TestToFlingerAnnounceDefault(t *testing.T) {
c := Config{}
if !c.ToFlinger().Announce {
t.Error("ToFlinger announce default should be true")
}
}
+45 -9
View File
@@ -48,23 +48,48 @@ type Config struct {
Audio bool
// StreamIndex selects which monitor to capture (screen source only).
StreamIndex int
// Announce controls whether the stream is advertised via UDP multicast.
// When disabled, receivers must connect by IP manually.
Announce bool
}
// DefaultConfig returns the recommended defaults.
func DefaultConfig() Config {
return Config{
Port: 9756,
Quality: 80,
FPS: 30,
Source: "screen",
Audio: true,
Port: 9756,
Quality: 80,
FPS: 30,
Source: "screen",
Audio: true,
Announce: true,
}
}
// Validate checks the config for out-of-range or unsupported values.
func (c Config) Validate() error {
if c.Port < 1 || c.Port > 65535 {
return errors.New("port must be 165535")
}
if c.Quality < 1 || c.Quality > 100 {
return errors.New("quality must be 1100")
}
if c.FPS < 1 || c.FPS > 240 {
return errors.New("fps must be 1240")
}
if c.Source != "screen" && c.Source != "pattern" {
return errors.New("source must be screen or pattern")
}
if c.StreamIndex < 0 {
return errors.New("stream index must be >= 0")
}
return nil
}
// Status is a point-in-time snapshot of the running engine.
type Status struct {
Running bool
Frames int64
Dropped int64
Conns int
}
@@ -87,6 +112,10 @@ type Engine struct {
// New creates an engine from cfg. Capture is opened eagerly so that
// misconfiguration (e.g. no screen-share permission) surfaces before Start.
func New(cfg Config) (*Engine, error) {
if err := cfg.Validate(); err != nil {
return nil, err
}
e := &Engine{cfg: cfg}
sender := output.New()
@@ -124,8 +153,8 @@ func New(cfg Config) (*Engine, error) {
return e, nil
}
// Start begins the audio, video and stats loops and starts announcing the
// stream. It is idempotent.
// Start begins the audio, video and stats loops and (unless disabled in the
// config) starts announcing the stream. It is idempotent.
func (e *Engine) Start() {
if e.stop != nil {
return
@@ -134,7 +163,9 @@ func (e *Engine) Start() {
e.start = time.Now()
e.stop = make(chan struct{})
e.announcer = discovery.Start(e.cfg.Name, e.sender.Port())
if e.cfg.Announce {
e.announcer = discovery.Start(e.cfg.Name, e.sender.Port())
}
var src io.ReadCloser
switch {
@@ -180,6 +211,7 @@ func (e *Engine) Status() Status {
return Status{
Running: e.stop != nil,
Frames: e.frames.Load(),
Dropped: e.sender.Dropped(),
Conns: e.sender.NumConns(),
}
}
@@ -276,7 +308,11 @@ func (e *Engine) statsLoop() {
select {
case <-tick.C:
st := e.Status()
log.Printf("flinger: %d frames, %d conns", st.Frames, st.Conns)
if st.Dropped > 0 {
log.Printf("flinger: %d frames, %d dropped, %d conns", st.Frames, st.Dropped, st.Conns)
} else {
log.Printf("flinger: %d frames, %d conns", st.Frames, st.Conns)
}
case <-e.stop:
return
}
+24 -1
View File
@@ -11,7 +11,7 @@ import (
func TestEnginePatternStartStop(t *testing.T) {
cfg := DefaultConfig()
cfg.Source = "pattern"
cfg.Port = 0 // ephemeral
cfg.Port = 19756 // fixed high port for the test
eng, err := New(cfg)
if err != nil {
@@ -77,6 +77,29 @@ func TestNewRejectsBadSource(t *testing.T) {
}
}
// TestValidate rejects out-of-range values.
func TestValidate(t *testing.T) {
bad := []func(*Config){
func(c *Config) { c.Port = 0 },
func(c *Config) { c.Port = 70000 },
func(c *Config) { c.Quality = 0 },
func(c *Config) { c.Quality = 101 },
func(c *Config) { c.FPS = 0 },
func(c *Config) { c.StreamIndex = -1 },
}
for i, mutate := range bad {
c := DefaultConfig()
mutate(&c)
if err := c.Validate(); err == nil {
t.Errorf("case %d: expected validation error", i)
}
}
if err := DefaultConfig().Validate(); err != nil {
t.Errorf("default config should validate: %v", err)
}
}
func itoa(v int) string {
if v == 0 {
return "0"
+58 -8
View File
@@ -10,6 +10,7 @@ import (
"errors"
"log"
"strconv"
"time"
"fyne.io/fyne/v2"
"fyne.io/fyne/v2/app"
@@ -43,14 +44,16 @@ type App struct {
lock bool // serialises start/stop against UI actions
// UI state.
startBtn *widget.Button
statusLab *widget.Label
portEnt *widget.Entry
qualitySel *widget.Select
fpsSel *widget.Select
nameEnt *widget.Entry
audioChk *widget.Check
srcSel *widget.Select
startBtn *widget.Button
statusLab *widget.Label
portEnt *widget.Entry
qualitySel *widget.Select
fpsSel *widget.Select
nameEnt *widget.Entry
audioChk *widget.Check
announceChk *widget.Check
srcSel *widget.Select
statsDone chan struct{}
}
// Run starts the GUI and blocks until the app exits.
@@ -120,6 +123,14 @@ func (g *App) buildUI() {
g.audioChk = widget.NewCheck("Capture system audio", nil)
g.audioChk.SetChecked(g.cfg.Audio)
// Announce over multicast.
g.announceChk = widget.NewCheck("Announce on LAN", nil)
announce := true
if g.cfg.Announce != nil {
announce = *g.cfg.Announce
}
g.announceChk.SetChecked(announce)
// Status label.
g.statusLab = widget.NewLabel("Stopped")
g.statusLab.Importance = widget.MediumImportance
@@ -136,6 +147,7 @@ func (g *App) buildUI() {
{Text: "Quality", Widget: g.qualitySel},
{Text: "Frame rate", Widget: g.fpsSel},
{Text: "", Widget: g.audioChk},
{Text: "", Widget: g.announceChk},
},
}
@@ -168,6 +180,7 @@ func (g *App) start() {
quality, _ := atoi(g.qualitySel.Selected)
fps, _ := atoi(g.fpsSel.Selected)
announce := g.announceChk.Checked
g.cfg = config.Config{
Name: g.nameEnt.Text,
Port: port,
@@ -176,6 +189,7 @@ func (g *App) start() {
Source: g.srcSel.Selected,
Audio: g.audioChk.Checked,
StreamIndex: g.cfg.StreamIndex,
Announce: &announce,
}
if err := config.Save(g.cfg); err != nil {
log.Printf("gui: config save: %v", err)
@@ -194,10 +208,46 @@ func (g *App) start() {
g.statusLab.SetText("Streaming (port " + itoa(g.cfg.Port) + ")")
g.statusLab.Importance = widget.SuccessImportance
g.refresh()
g.watchStats()
}
// watchStats refreshes the status label with live counters while streaming.
func (g *App) watchStats() {
done := make(chan struct{})
g.statsDone = done
go func() {
tick := time.NewTicker(2 * time.Second)
defer tick.Stop()
for {
select {
case <-tick.C:
if g.eng == nil {
return
}
st := g.eng.Status()
g.statusLab.SetText(formatStatus(st))
case <-done:
return
}
}
}()
}
// formatStatus renders the live status line.
func formatStatus(st flinger.Status) string {
base := "Streaming · " + itoa(int(st.Frames)) + " frames"
if st.Dropped > 0 {
base += " · " + itoa(int(st.Dropped)) + " dropped"
}
return base
}
// stop halts the engine and returns the UI to the stopped state.
func (g *App) stop() {
if g.statsDone != nil {
close(g.statsDone)
g.statsDone = nil
}
if g.eng != nil {
g.eng.Stop()
g.eng = nil
+12 -1
View File
@@ -16,6 +16,7 @@ import (
"log"
"net"
"sync"
"sync/atomic"
)
const (
@@ -34,6 +35,8 @@ type Sender struct {
listener net.Listener
port int
dropped atomic.Int64
}
// New creates an unconnected Sender.
@@ -86,7 +89,8 @@ func (s *Sender) NumConns() int {
// Send broadcasts a serialised packet to all connected receivers. If a
// receiver's buffered channel is full (> dropAt) the frame is silently
// dropped; a warning is logged at warnAt.
// dropped; a warning is logged at warnAt. Dropped frames are counted and
// exposed via Dropped.
func (s *Sender) Send(b []byte) {
s.mu.Lock()
defer s.mu.Unlock()
@@ -101,6 +105,7 @@ func (s *Sender) Send(b []byte) {
}
if len(ch) > dropAt {
s.dropped.Add(1)
continue
}
@@ -109,11 +114,17 @@ func (s *Sender) Send(b []byte) {
select {
case ch <- b:
default:
s.dropped.Add(1)
log.Printf("output: drop [%s] (queue full)", c.RemoteAddr())
}
}
}
// Dropped returns the total number of frames dropped due to backpressure.
func (s *Sender) Dropped() int64 {
return s.dropped.Load()
}
// Close shuts down the listener and waits for all writer goroutines to
// drain. After Close returns the Sender must not be reused.
func (s *Sender) Close() {
+21
View File
@@ -149,3 +149,24 @@ Components:
for now (loopback/LAN testing). A remote receiver machine can be added later for
further testing.
6. **License**: GPL-2.0, matching the `obs-teleport` project whose protocol we implement.
## Decisions Made (2026-09-18, M3/M4)
7. **Screen capture backend**: `go2tv.app/screencast` (MIT, GPL-2.0-compatible) via
`xdg-desktop-portal` + PipeWire. Chosen over hand-rolled cgo bindings. The portal
path is compositor-agnostic: Wayland (hyprland/gnome portals) and X11
(`xdg-desktop-portal-gtk`) both work, so a separate native X11/XShm fallback is not
required for the supported path and was not implemented (project.md M4 "if feasible").
8. **GUI toolkit**: Fyne v2 — provides both the settings window and a system tray
(StatusNotifierItem over DBus), so no separate systray dependency was needed.
9. **Engine decoupling**: streaming core lives in `internal/flinger` (GUI-free) so the
CLI, GUI and any future daemon share one implementation.
## M4 Hardening Status
- Config persistence + validation: done (internal/config, flinger.Validate).
- Backpressure/queue stats: done (sender tracks dropped frames; exposed in engine
Status and the GUI status label).
- Discovery on/off: done (flinger.Config.Announce, GUI checkbox).
- X11 fallback: not implemented — the portal backend already covers X11 sessions.
- Packaging (Nix package / AppImage): deferred (see AGENTS.md / roadmap).