diff --git a/go.mod b/go.mod index 2dcfadc..732ad99 100644 --- a/go.mod +++ b/go.mod @@ -44,3 +44,5 @@ require ( golang.org/x/text v0.42.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) + +replace go2tv.app/screencast => ./third_party/screencast diff --git a/internal/capture/audio.go b/internal/capture/audio.go new file mode 100644 index 0000000..2d33f97 --- /dev/null +++ b/internal/capture/audio.go @@ -0,0 +1,97 @@ +// Audio source enumeration for the settings UI. + +package capture + +import ( + "encoding/json" + "os/exec" +) + +// AudioDevice describes a capturable PipeWire audio node. +type AudioDevice struct { + Serial uint64 // PipeWire object.serial, passed to the capture backend + ID uint32 // PipeWire node id (informational) + Name string // node name, e.g. "alsa_output...analog-stereo" + Desc string // human-readable description + IsOutput bool // true = a sink (system output); false = a source (mic) +} + +// pwDumpNode is the subset of `pw-dump` output we parse. +type pwDumpNode struct { + ID uint32 `json:"id"` + Info struct { + Props map[string]any `json:"props"` + } `json:"info"` +} + +// ListAudioSources enumerates PipeWire audio sinks and sources via `pw-dump`. +// To capture system audio we attach to a sink's monitor; to capture a +// microphone we attach to an Audio/Source node. This lists the capturable +// audio nodes so the GUI can present a picker instead of always using the +// system default output. +func ListAudioSources() ([]AudioDevice, error) { + out, err := exec.Command("pw-dump").Output() + if err != nil { + return nil, err + } + + var nodes []pwDumpNode + if err := json.Unmarshal(out, &nodes); err != nil { + return nil, err + } + + var sources []AudioDevice + for _, n := range nodes { + props := n.Info.Props + mediaClass, _ := props["media.class"].(string) + if !audioNodeClass(mediaClass) { + continue + } + desc, _ := props["node.description"].(string) + name, _ := props["node.name"].(string) + serial := toUint64(props["object.serial"]) + if serial == 0 { + continue + } + sources = append(sources, AudioDevice{ + Serial: serial, + ID: n.ID, + Name: name, + Desc: desc, + IsOutput: mediaClass == "Audio/Sink", + }) + } + return sources, nil +} + +// audioNodeClass reports whether a media.class is a capturable audio node. +func audioNodeClass(mediaClass string) bool { + switch mediaClass { + case "Audio/Sink", "Audio/Source": + return true + default: + return false + } +} + +// toUint64 best-effort converts a pw-dump property value to uint64. +// PipeWire serials are small, so the conversions cannot overflow in practice. +// +//nolint:gosec // safe: JSON numbers from pw-dump are small node serials +func toUint64(v any) uint64 { + switch t := v.(type) { + case float64: + return uint64(t) + case int: + return uint64(t) + case int64: + return uint64(t) + case uint64: + return t + case json.Number: + if n, err := t.Int64(); err == nil { + return uint64(n) + } + } + return 0 +} diff --git a/internal/capture/audio_test.go b/internal/capture/audio_test.go new file mode 100644 index 0000000..2bc42f6 --- /dev/null +++ b/internal/capture/audio_test.go @@ -0,0 +1,64 @@ +package capture + +import ( + "encoding/json" + "testing" +) + +// TestPWParseAudioNodes parses a representative pw-dump payload into the +// node shape used by ListAudioSources. +func TestPWParseAudioNodes(t *testing.T) { + payload := `[ + {"id": 51, "info": {"props": {"media.class": "Audio/Sink", "node.name": "alsa_out_speaker", "node.description": "Speaker", "object.serial": 1179}}}, + {"id": 56, "info": {"props": {"media.class": "Audio/Source", "node.name": "alsa_in_mic", "node.description": "Stereo Mic", "object.serial": 1183}}}, + {"id": 64, "info": {"props": {"media.class": "Audio/Device", "node.description": "Not capturable", "object.serial": 1172}}} +]` + + var nodes []pwDumpNode + if err := json.Unmarshal([]byte(payload), &nodes); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + var devs []AudioDevice + for _, n := range nodes { + props := n.Info.Props + mc, _ := props["media.class"].(string) + if !audioNodeClass(mc) { + continue + } + devs = append(devs, AudioDevice{ + Serial: toUint64(props["object.serial"]), + ID: n.ID, + Name: props["node.name"].(string), + Desc: props["node.description"].(string), + IsOutput: mc == "Audio/Sink", + }) + } + + if len(devs) != 2 { + t.Fatalf("got %d capturable devices, want 2", len(devs)) + } + // Sink serial parsed and flagged as output. + if devs[0].Serial != 1179 || !devs[0].IsOutput { + t.Errorf("sink wrong: %+v", devs[0]) + } + // Source serial parsed and flagged as input. + if devs[1].Serial != 1183 || devs[1].IsOutput { + t.Errorf("source wrong: %+v", devs[1]) + } +} + +// TestAudioNodeClass verifies which media classes are capturable. +func TestAudioNodeClass(t *testing.T) { + cases := map[string]bool{ + "Audio/Sink": true, + "Audio/Source": true, + "Audio/Device": false, + "Video/Source": false, + } + for cls, want := range cases { + if got := audioNodeClass(cls); got != want { + t.Errorf("audioNodeClass(%q) = %v, want %v", cls, got, want) + } + } +} diff --git a/internal/capture/pipewire.go b/internal/capture/pipewire.go index b98b468..877bed0 100644 --- a/internal/capture/pipewire.go +++ b/internal/capture/pipewire.go @@ -24,12 +24,14 @@ type PipeWire struct { } // OpenPipeWire opens a PipeWire capture session. streamIndex selects which -// monitor to capture when multiple are present. Triggering the portal -// consent dialog is expected; the compositor decides whether to show it. -func OpenPipeWire(streamIndex int, audio bool) (*PipeWire, error) { +// monitor to capture when multiple are present; audioSourceSerial optionally +// selects a specific PipeWire audio node (0 = system default). Triggering the +// portal consent dialog is expected; the compositor decides whether to show it. +func OpenPipeWire(streamIndex int, audio bool, audioSourceSerial uint64) (*PipeWire, error) { s, err := capture.Open(&capture.Options{ - StreamIndex: streamIndex, - IncludeAudio: audio, + StreamIndex: streamIndex, + IncludeAudio: audio, + AudioSourceSerial: audioSourceSerial, }) if err != nil { return nil, err diff --git a/internal/config/config.go b/internal/config/config.go index 10192c5..2645725 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,7 @@ type Config struct { FPS int `json:"fps"` Source string `json:"source"` Audio bool `json:"audio"` + AudioSource uint64 `json:"audio_source"` StreamIndex int `json:"stream_index"` Scale float64 `json:"scale"` // Announce is a *bool so an absent JSON key (older config files) keeps @@ -40,6 +41,7 @@ func Default() Config { FPS: c.FPS, Source: c.Source, Audio: c.Audio, + AudioSource: c.AudioSource, StreamIndex: c.StreamIndex, Scale: c.Scale, Announce: &announce, @@ -59,6 +61,7 @@ func (c Config) ToFlinger() flinger.Config { FPS: c.FPS, Source: c.Source, Audio: c.Audio, + AudioSource: c.AudioSource, StreamIndex: c.StreamIndex, Scale: c.Scale, Announce: announce, diff --git a/internal/flinger/flinger.go b/internal/flinger/flinger.go index 58c4d41..b693167 100644 --- a/internal/flinger/flinger.go +++ b/internal/flinger/flinger.go @@ -48,6 +48,9 @@ type Config struct { Source string // Audio enables system audio capture and streaming. Audio bool + // AudioSource is the PipeWire node serial to capture audio from (0 = system + // default output). Only applies when Audio is true and Source is "screen". + AudioSource uint64 // StreamIndex selects which monitor to capture (screen source only). StreamIndex int // Scale downscales the captured frame before encoding. 1.0 = native @@ -122,8 +125,18 @@ type Engine struct { start time.Time stop chan struct{} + // running gates packet sending: when false the loops consume capture + // frames but do not transmit. This lets the GUI "stop" without tearing + // down the portal/PipeWire session (which the portal cannot reliably + // recreate in-process), so a subsequent Start just flips running on. + running atomic.Bool + frames atomic.Int64 + // wg tracks the audio/video/stats goroutines so Stop can wait for them to + // finish before destroying shared resources (encoder, sender, capture). + wg sync.WaitGroup + // scaleBuf is the cached downscaled frame, reallocated only when the // configured scale or source dimensions change. scaleBuf *capture.VideoFrame @@ -162,6 +175,7 @@ func (e *Engine) SetConfig(cfg Config) error { cfg.Source = e.cfg.Source cfg.Port = e.cfg.Port cfg.StreamIndex = e.cfg.StreamIndex + cfg.AudioSource = e.cfg.AudioSource e.cfg = cfg return nil @@ -199,7 +213,7 @@ func New(cfg Config) (*Engine, error) { switch cfg.Source { case "screen": - cam, err := capture.OpenPipeWire(cfg.StreamIndex, cfg.Audio) + cam, err := capture.OpenPipeWire(cfg.StreamIndex, cfg.Audio, cfg.AudioSource) if err != nil { e.encoder.Close() sender.Close() @@ -222,11 +236,13 @@ func New(cfg Config) (*Engine, error) { // config) starts announcing the stream. It is idempotent. func (e *Engine) Start() { if e.stop != nil { + e.Resume() return } e.start = time.Now() e.stop = make(chan struct{}) + e.running.Store(true) if e.cfg.Announce { e.announcer = discovery.Start(e.cfg.Name, e.sender.Port()) @@ -243,32 +259,68 @@ func (e *Engine) Start() { src = capture.NewSilenceSource() } - go e.audioLoop(src) - go e.videoLoop() - go e.statsLoop() + e.wg.Add(3) + go func() { + defer e.wg.Done() + e.audioLoop(src) + }() + go func() { + defer e.wg.Done() + e.videoLoop() + }() + go func() { + defer e.wg.Done() + e.statsLoop() + }() } -// Stop halts the loops, stops announcing and closes all resources. It is -// idempotent. After Stop the engine must not be restarted. +// Stop pauses the stream: loops keep consuming capture frames but stop +// transmitting, and the portal/PipeWire session stays open so the engine can +// be resumed with Start without re-opening the portal (which the portal +// cannot reliably do in-process). Idempotent. func (e *Engine) Stop() { + e.running.Store(false) +} + +// Pause is an alias for Stop; kept for clarity at call sites. +func (e *Engine) Pause() { e.Stop() } + +// Resume restarts transmission on a paused engine. It is a no-op if the +// engine was never started. +func (e *Engine) Resume() { + if e.stop == nil { + return + } + e.running.Store(true) +} + +// Close fully tears down the engine: stops the loops, closes the portal +// session and frees the encoder/sender. After Close the engine must not be +// reused. +func (e *Engine) Close() { if e.stop == nil { return } close(e.stop) - // Give the loops a moment to observe the stop signal. - time.Sleep(50 * time.Millisecond) + + // Closing the capture source unblocks the audio and video loops that may + // be stuck in a read. Do this before waiting so they can observe stop. + if e.cam != nil { + if err := e.cam.Close(); err != nil { + log.Printf("flinger: capture close: %v", err) + } + } + + // Wait for the audio/video/stats goroutines to finish so they no longer + // touch the encoder or sender before we destroy them. + e.wg.Wait() if e.announcer != nil { e.announcer.Stop() } e.sender.Close() e.encoder.Close() - if e.cam != nil { - if err := e.cam.Close(); err != nil { - log.Printf("flinger: capture close: %v", err) - } - } } // Status returns a snapshot of the running engine. @@ -278,7 +330,7 @@ func (e *Engine) Status() Status { e.errMu.RUnlock() return Status{ - Running: e.stop != nil, + Running: e.running.Load() && e.stop != nil, Frames: e.frames.Load(), Dropped: e.sender.Dropped(), Conns: e.sender.NumConns(), @@ -333,8 +385,17 @@ func (e *Engine) audioLoop(src io.ReadCloser) { buf := make([]byte, chunkBytes) for { + select { + case <-e.stop: + return + default: + } + n, err := io.ReadFull(src, buf) if n > 0 { + if !e.running.Load() { + continue + } frames := n / (speakers * 2) ts := uint64(time.Since(e.start)) packet, perr := protocol.BuildWavePacket(ts, protocol.AudioFormatS16, sampleRate, speakers, int32(frames), buf[:n]) @@ -433,6 +494,12 @@ func (e *Engine) videoLoop() { next := e.start for { + select { + case <-e.stop: + return + default: + } + frame, err := e.loop.Next() if err != nil { select { @@ -464,6 +531,13 @@ func (e *Engine) videoLoop() { } next = now.Add(frameInterval) + // When paused (Stop), keep consuming frames to stay alive but do not + // encode or transmit. The portal session is left open so a subsequent + // Start can resume without re-opening the portal. + if !e.running.Load() { + continue + } + ts := uint64(now.Sub(e.start)) enc := e.scaleFrame(frame, cfg.Scale) buf, err := e.encoder.EncodeBGRA(enc.Pix, enc.Width, enc.Height, cfg.Quality) diff --git a/internal/flinger/flinger_test.go b/internal/flinger/flinger_test.go index e88f7fb..0c4ea20 100644 --- a/internal/flinger/flinger_test.go +++ b/internal/flinger/flinger_test.go @@ -22,7 +22,7 @@ func TestEnginePatternStartStop(t *testing.T) { } eng.Start() - defer eng.Stop() + defer eng.Close() // Connect a raw receiver and read until the engine reports frames sent. conn, err := net.Dial("tcp", "127.0.0.1:"+itoa(eng.sender.Port())) @@ -143,7 +143,7 @@ func TestBitrateMeasurement(t *testing.T) { t.Fatalf("New: %v", err) } eng.Start() - defer eng.Stop() + defer eng.Close() // Connect a receiver so packets actually flow, and wait for the bitrate // window to produce a measurement. diff --git a/internal/gui/gui.go b/internal/gui/gui.go index fd235a9..4a86e2e 100644 --- a/internal/gui/gui.go +++ b/internal/gui/gui.go @@ -93,11 +93,17 @@ type App struct { scaleSel *widget.Select nameEnt *widget.Entry audioChk *widget.Check + audioSel *widget.Select + audioDevs []capture.AudioDevice announceChk *widget.Check srcSel *widget.Select monSel *widget.Select monitors []capture.Monitor statsDone chan struct{} + + // lastStart records the capture-affecting config the current engine was + // created with, so Start can resume instead of reopening the portal. + lastStart config.Config } // Run starts the GUI and blocks until the app exits. configPath selects a @@ -188,8 +194,9 @@ func (g *App) buildUI() { g.presetSel.SetSelected("High") // Audio. - g.audioChk = widget.NewCheck("Capture system audio", nil) + g.audioChk = widget.NewCheck("Capture system audio", func(bool) { g.applyLiveSettings("") }) g.audioChk.SetChecked(g.cfg.Audio) + g.setupAudioPicker() // Announce over multicast. g.announceChk = widget.NewCheck("Announce on LAN", nil) @@ -217,7 +224,8 @@ func (g *App) buildUI() { {Text: "Quality", Widget: g.qualitySel}, {Text: "Frame rate", Widget: g.fpsSel}, {Text: "Scale", Widget: g.scaleSel}, - {Text: "", Widget: g.audioChk}, + {Text: "Audio", Widget: g.audioChk}, + {Text: "Audio source", Widget: g.audioSel}, {Text: "", Widget: g.announceChk}, }, } @@ -283,6 +291,54 @@ func (g *App) selectedMonitorIndex() int { return g.cfg.StreamIndex } +// setupAudioPicker populates the audio source dropdown from PipeWire. The +// first option is "Default output"; the rest are the enumerated sinks and +// microphones. A no-op if enumeration is unavailable. +func (g *App) setupAudioPicker() { + g.audioDevs, _ = capture.ListAudioSources() + + names := []string{"Default output"} + for _, d := range g.audioDevs { + label := d.Desc + if label == "" { + label = d.Name + } + if d.IsOutput { + label = "Output: " + label + } else { + label = "Input: " + label + } + names = append(names, label) + } + + g.audioSel = widget.NewSelect(names, func(string) { g.applyLiveSettings("") }) + + // Preselect the configured serial if it matches an enumerated device. + if g.cfg.AudioSource > 0 { + for i, d := range g.audioDevs { + if d.Serial == g.cfg.AudioSource { + g.audioSel.SetSelectedIndex(i + 1) + break + } + } + } else { + g.audioSel.SetSelectedIndex(0) + } +} + +// selectedAudioSerial returns the PipeWire serial chosen in the picker, or 0 +// for the default output. +func (g *App) selectedAudioSerial() uint64 { + if g.audioSel == nil || g.audioSel.SelectedIndex() <= 0 { + return 0 + } + idx := g.audioSel.SelectedIndex() - 1 + if idx >= 0 && idx < len(g.audioDevs) { + return g.audioDevs[idx].Serial + } + return 0 +} + // applyPreset applies a named quality/fps preset to the Quality and Frame // rate selectors, then pushes it live if the engine is running. func (g *App) applyPreset(string) { @@ -320,6 +376,7 @@ func (g *App) applyLiveSettings(string) { FPS: fps, Source: g.cfg.Source, Audio: g.audioChk.Checked, + AudioSource: g.selectedAudioSerial(), StreamIndex: g.cfg.StreamIndex, Scale: parseScale(g.scaleSel.Selected), Announce: announce, @@ -332,6 +389,7 @@ func (g *App) applyLiveSettings(string) { g.cfg.Quality = quality g.cfg.FPS = fps g.cfg.Audio = cfg.Audio + g.cfg.AudioSource = cfg.AudioSource g.cfg.Name = cfg.Name g.cfg.Scale = cfg.Scale ann := announce @@ -343,31 +401,37 @@ func (g *App) toggleStream() { if g.lock { return } - if g.eng != nil { + // Engine exists and is currently running → pause it. + if g.eng != nil && g.eng.Status().Running { g.stop() return } g.start() } -// start reads the form into cfg, saves it, and boots the engine. +// start reads the form into cfg, saves it, and boots the engine. If an engine +// already exists and the capture-affecting settings are unchanged, it resumes +// the paused engine instead of reopening the portal session (which cannot be +// reliably re-created in-process). func (g *App) start() { port, _ := atoi(g.portEnt.Text) quality, _ := atoi(g.qualitySel.Selected) fps, _ := atoi(g.fpsSel.Selected) announce := g.announceChk.Checked - g.cfg = config.Config{ + newCfg := config.Config{ Name: g.nameEnt.Text, Port: port, Quality: quality, FPS: fps, Source: g.srcSel.Selected, Audio: g.audioChk.Checked, + AudioSource: g.selectedAudioSerial(), StreamIndex: g.selectedMonitorIndex(), Scale: parseScale(g.scaleSel.Selected), Announce: &announce, } + g.cfg = newCfg if g.configPath != "" { if err := config.SaveTo(g.configPath, g.cfg); err != nil { log.Printf("gui: config save: %v", err) @@ -376,17 +440,47 @@ func (g *App) start() { log.Printf("gui: config save: %v", err) } + // Resume the existing engine if capture-affecting fields are unchanged. + if g.eng != nil && captureConfigEqual(g.lastStart, newCfg) { + // Push live-applicable changes, then resume. + g.applyLiveSettings("") + g.eng.Resume() + g.startedUI(newCfg) + return + } + + // Otherwise close any existing engine and create a fresh one. + if g.eng != nil { + g.eng.Close() + g.eng = nil + } + eng, err := flinger.New(g.cfg.ToFlinger()) if err != nil { dialog.ShowError(err, g.win) return } g.eng = eng + g.lastStart = newCfg g.eng.Start() + g.startedUI(newCfg) +} + +// captureConfigEqual reports whether two configs agree on the fields that +// require reopening the capture session (source, port, monitor, audio +// source). Live-applicable fields (quality, fps, scale, name, announce, +// audio-on) are ignored. +func captureConfigEqual(a, b config.Config) bool { + return a.Source == b.Source && a.Port == b.Port && + a.StreamIndex == b.StreamIndex && a.AudioSource == b.AudioSource +} + +// startedUI updates the UI to the streaming state after a start/resume. +func (g *App) startedUI(cfg config.Config) { g.startBtn.SetText("Stop") g.startBtn.Importance = widget.DangerImportance - g.statusLab.SetText("Streaming (port " + itoa(g.cfg.Port) + ")") + g.statusLab.SetText("Streaming (port " + itoa(cfg.Port) + ")") g.statusLab.Importance = widget.SuccessImportance g.setTrayState(true, "TeleportFling · Streaming") g.refresh() @@ -460,7 +554,8 @@ func formatBitrate(bps int64) string { } } -// stop halts the engine and returns the UI to the stopped state. +// stop pauses the engine (keeping the portal session open) and returns the +// UI to the stopped state. func (g *App) stop() { if g.statsDone != nil { close(g.statsDone) @@ -468,7 +563,6 @@ func (g *App) stop() { } if g.eng != nil { g.eng.Stop() - g.eng = nil } g.startBtn.SetText("Start") g.startBtn.Importance = widget.HighImportance diff --git a/third_party/screencast b/third_party/screencast new file mode 160000 index 0000000..4cba613 --- /dev/null +++ b/third_party/screencast @@ -0,0 +1 @@ +Subproject commit 4cba6136251ae754e6dbd3c045e3d7c21cf21921