refactor: extract streaming engine into internal/flinger

Move the capture/encode/send pipeline out of cmd/teleportfling into a
GUI-free Engine so the CLI and the desktop app share one implementation:
- flinger: Config/Engine/Status, audio+video+stats loops, pattern source
- config: JSON persistence at ~/.config/teleportfling/config.json
- cmd/teleportfling: thin CLI wrapper around the engine (same flags)
- golangci: extend exclusions to flinger/config
This commit is contained in:
2026-09-18 20:32:45 +01:00
parent 79beeefdc7
commit ba423eb944
8 changed files with 836 additions and 368 deletions
+113
View File
@@ -0,0 +1,113 @@
// Package config loads and saves the teleportfling settings file.
//
// The file is JSON at ~/.config/teleportfling/config.json and holds the
// user-visible knobs exposed by the GUI (name, port, quality, fps, source,
// audio, monitor). Secrets and runtime state are deliberately excluded.
package config
import (
"encoding/json"
"errors"
"os"
"path/filepath"
"teleportfling/internal/flinger"
)
// Config mirrors flinger.Config with JSON tags for persistence.
type Config struct {
Name string `json:"name"`
Port int `json:"port"`
Quality int `json:"quality"`
FPS int `json:"fps"`
Source string `json:"source"`
Audio bool `json:"audio"`
StreamIndex int `json:"stream_index"`
}
// Default returns the default configuration.
func Default() Config {
c := flinger.DefaultConfig()
return Config{
Name: c.Name,
Port: c.Port,
Quality: c.Quality,
FPS: c.FPS,
Source: c.Source,
Audio: c.Audio,
StreamIndex: c.StreamIndex,
}
}
// ToFlinger converts a persisted config to the engine config.
func (c Config) ToFlinger() flinger.Config {
return flinger.Config{
Name: c.Name,
Port: c.Port,
Quality: c.Quality,
FPS: c.FPS,
Source: c.Source,
Audio: c.Audio,
StreamIndex: c.StreamIndex,
}
}
// pathVar is the config file location; overridable in tests.
var pathVar = func() string {
dir, err := os.UserConfigDir()
if err != nil {
dir = "."
}
return filepath.Join(dir, "teleportfling", "config.json")
}()
// Path returns the config file location.
func Path() string {
return pathVar
}
// Load reads the config file, returning Default when it does not exist.
func Load() (Config, error) {
p := Path()
data, err := os.ReadFile(p)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return Default(), nil
}
return Config{}, err
}
var c Config
if err := json.Unmarshal(data, &c); err != nil {
return Config{}, err
}
// Fill any zero values with defaults so a hand-edited file still works.
d := Default()
if c.Port == 0 {
c.Port = d.Port
}
if c.Quality == 0 {
c.Quality = d.Quality
}
if c.FPS == 0 {
c.FPS = d.FPS
}
if c.Source == "" {
c.Source = d.Source
}
return c, nil
}
// Save writes the config file, creating the directory if needed.
func Save(c Config) error {
p := Path()
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
return err
}
data, err := json.MarshalIndent(c, "", " ")
if err != nil {
return err
}
return os.WriteFile(p, data, 0o600)
}
+77
View File
@@ -0,0 +1,77 @@
package config
import (
"os"
"path/filepath"
"testing"
)
// TestLoadDefaultsWhenMissing verifies Load returns defaults when the file
// does not exist.
func TestLoadDefaultsWhenMissing(t *testing.T) {
// Point at a non-existent directory so we never touch the real config.
old := pathVar
pathVar = filepath.Join(t.TempDir(), "nope", "config.json")
defer func() { pathVar = old }()
c, err := Load()
if err != nil {
t.Fatalf("Load: %v", err)
}
if c.Port != 9756 {
t.Errorf("default port = %d, want 9756", c.Port)
}
}
// TestSaveLoadRoundTrip writes a config and reads it back.
func TestSaveLoadRoundTrip(t *testing.T) {
old := pathVar
pathVar = filepath.Join(t.TempDir(), "config.json")
defer func() { pathVar = old }()
want := Config{
Name: "Studio",
Port: 9898,
Quality: 95,
FPS: 60,
Source: "pattern",
Audio: false,
StreamIndex: 1,
}
if err := Save(want); err != nil {
t.Fatalf("Save: %v", err)
}
got, err := Load()
if err != nil {
t.Fatalf("Load: %v", err)
}
if got != want {
t.Errorf("round trip mismatch:\n got %+v\nwant %+v", got, want)
}
}
// TestLoadFillsZeroValues ensures a partial file gets defaults filled.
func TestLoadFillsZeroValues(t *testing.T) {
old := pathVar
pathVar = filepath.Join(t.TempDir(), "config.json")
defer func() { pathVar = old }()
if err := os.WriteFile(pathVar, []byte(`{"name":"Partial"}`), 0o600); err != nil {
t.Fatalf("write: %v", err)
}
c, err := Load()
if err != nil {
t.Fatalf("Load: %v", err)
}
if c.Name != "Partial" {
t.Errorf("name = %q, want Partial", c.Name)
}
if c.Port != 9756 {
t.Errorf("port = %d, want filled default 9756", c.Port)
}
if c.Quality != 80 {
t.Errorf("quality = %d, want filled default 80", c.Quality)
}
}
+423
View File
@@ -0,0 +1,423 @@
// Package flinger implements the TeleportFling streaming engine: it captures
// a screen (and optionally system audio), encodes video to JPEG, packetizes
// both into the Teleport protocol, and broadcasts them to connected OBS
// receivers.
//
// The engine is deliberately GUI-free so it can run headless (CLI/daemon) or
// be driven by a desktop app. Callers create an Engine with a Config, call
// Start, poll Status, and call Stop when done.
package flinger
import (
"errors"
"image"
"image/color"
"io"
"log"
"strconv"
"sync/atomic"
"time"
"teleportfling/internal/capture"
"teleportfling/internal/discovery"
"teleportfling/internal/output"
"teleportfling/internal/protocol"
)
const (
// sampleRate and speakers describe the captured/encoded audio stream.
sampleRate = 48000
speakers = 2
// audioChunk sets how much audio we packetize per WAVE message (~10 ms).
audioChunk = 10 * time.Millisecond
)
// Config configures the streaming engine.
type Config struct {
// Name is the announce name advertised to receivers (empty → hostname).
Name string
// Port is the TCP listening port for receivers.
Port int
// Quality is the JPEG quality, 1100.
Quality int
// FPS is the target video frame rate.
FPS int
// Source is "screen" (PipeWire) or "pattern" (synthetic test signal).
Source string
// Audio enables system audio capture and streaming.
Audio bool
// StreamIndex selects which monitor to capture (screen source only).
StreamIndex int
}
// DefaultConfig returns the recommended defaults.
func DefaultConfig() Config {
return Config{
Port: 9756,
Quality: 80,
FPS: 30,
Source: "screen",
Audio: true,
}
}
// Status is a point-in-time snapshot of the running engine.
type Status struct {
Running bool
Frames int64
Conns int
}
// Engine owns the capture, encode and send pipeline.
type Engine struct {
cfg Config
sender *output.Sender
announcer *discovery.Announcer
encoder *protocol.JPEGEncoder
cam capture.Capture
loop frameSource
start time.Time
stop chan struct{}
frames atomic.Int64
}
// 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) {
e := &Engine{cfg: cfg}
sender := output.New()
if _, err := sender.Listen(addr(cfg.Port)); err != nil {
return nil, err
}
enc, err := protocol.NewJPEGEncoder()
if err != nil {
sender.Close()
return nil, err
}
e.sender = sender
e.encoder = enc
switch cfg.Source {
case "screen":
cam, err := capture.OpenPipeWire(cfg.StreamIndex, cfg.Audio)
if err != nil {
e.encoder.Close()
sender.Close()
return nil, err
}
e.cam = cam
e.loop = captureLoop{cam}
case "pattern":
e.loop = &patternLoop{w: 1920, h: 1080}
default:
e.encoder.Close()
sender.Close()
return nil, errors.New("unknown source " + cfg.Source)
}
return e, nil
}
// Start begins the audio, video and stats loops and starts announcing the
// stream. It is idempotent.
func (e *Engine) Start() {
if e.stop != nil {
return
}
e.start = time.Now()
e.stop = make(chan struct{})
e.announcer = discovery.Start(e.cfg.Name, e.sender.Port())
var src io.ReadCloser
switch {
case e.cam == nil:
src = capture.NewSilenceSource()
case e.cam.Audio() != nil:
src = e.cam.Audio()
default:
log.Printf("flinger: system audio unavailable, streaming silence")
src = capture.NewSilenceSource()
}
go e.audioLoop(src)
go e.videoLoop()
go e.statsLoop()
}
// Stop halts the loops, stops announcing and closes all resources. It is
// idempotent. After Stop the engine must not be restarted.
func (e *Engine) Stop() {
if e.stop == nil {
return
}
close(e.stop)
// Give the loops a moment to observe the stop signal.
time.Sleep(50 * time.Millisecond)
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.
func (e *Engine) Status() Status {
return Status{
Running: e.stop != nil,
Frames: e.frames.Load(),
Conns: e.sender.NumConns(),
}
}
// audioLoop reads raw PCM and emits WAVE packets. start is the shared
// reference clock used by the video loop so audio and video timestamps stay
// aligned on the receiver.
func (e *Engine) audioLoop(src io.ReadCloser) {
defer func() { _ = src.Close() }()
framesPerChunk := int(sampleRate) * int(audioChunk) / int(time.Second)
chunkBytes := framesPerChunk * speakers * 2 // S16
buf := make([]byte, chunkBytes)
for {
n, err := io.ReadFull(src, buf)
if n > 0 {
frames := n / (speakers * 2)
ts := uint64(time.Since(e.start))
packet, perr := protocol.BuildWavePacket(ts, protocol.AudioFormatS16, sampleRate, speakers, int32(frames), buf[:n])
if perr != nil {
log.Printf("flinger: wave: %v", perr)
} else {
e.sender.Send(packet)
}
}
if err != nil {
select {
case <-e.stop:
return
default:
}
if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrClosedPipe) {
log.Printf("flinger: audio: %v", err)
}
}
}
}
// videoLoop pulls frames and sends them at the configured fps.
func (e *Engine) videoLoop() {
frameInterval := time.Second / time.Duration(e.cfg.FPS)
next := e.start
for {
frame, err := e.loop.Next()
if err != nil {
select {
case <-e.stop:
return
default:
}
if !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrClosedPipe) {
log.Printf("flinger: capture: %v", err)
}
continue
}
// Drop frames when running ahead of the target fps to keep
// timestamps monotonic (e.g. a 60 Hz monitor captured at 30 fps).
now := time.Now()
if now.Before(next) {
continue
}
next = now.Add(frameInterval)
ts := uint64(now.Sub(e.start))
buf, err := e.encoder.EncodeBGRA(frame.Pix, frame.Width, frame.Height, e.cfg.Quality)
if err != nil {
log.Printf("flinger: jpeg: %v", err)
continue
}
packet, err := protocol.WritePacket(
protocol.Header{Type: protocol.VideoType, Timestamp: ts, Size: int32(len(buf))},
ptr(protocol.DefaultBT709Full()),
nil,
buf,
)
if err != nil {
log.Printf("flinger: packet: %v", err)
continue
}
e.sender.Send(packet)
e.frames.Add(1)
}
}
// statsLoop logs a periodic summary.
func (e *Engine) statsLoop() {
tick := time.NewTicker(5 * time.Second)
defer tick.Stop()
for {
select {
case <-tick.C:
st := e.Status()
log.Printf("flinger: %d frames, %d conns", st.Frames, st.Conns)
case <-e.stop:
return
}
}
}
// frameSource abstracts the frame source: real capture or the test pattern.
type frameSource interface {
Next() (*capture.VideoFrame, error)
}
// captureLoop wraps the PipeWire capture backend.
type captureLoop struct {
cam capture.Capture
}
func (c captureLoop) Next() (*capture.VideoFrame, error) {
return c.cam.Video().NextFrame()
}
// patternLoop synthesizes the M1 test pattern (colour bars + moving box).
type patternLoop struct {
w, h int
seq int64
}
func (p *patternLoop) Next() (*capture.VideoFrame, error) {
img := testPattern(p.w, p.h, int(p.seq))
p.seq++
return ycrcbToBGRA(img), nil
}
// ycrcbToBGRA converts a YCbCr image to a BGRA VideoFrame so both sources
// share the encode path (EncodeBGRA).
func ycrcbToBGRA(img *image.YCbCr) *capture.VideoFrame {
w, h := img.Rect.Dx(), img.Rect.Dy()
frame := &capture.VideoFrame{
Pix: make([]byte, w*h*4),
Width: w,
Height: h,
Stride: w * 4,
}
for y := 0; y < h; y++ {
for x := 0; x < w; x++ {
yi := y*img.YStride + x
ci := (y/2)*img.CStride + x/2
r, g, b := color.YCbCrToRGB(img.Y[yi], img.Cb[ci], img.Cr[ci])
off := (y*w + x) * 4
frame.Pix[off], frame.Pix[off+1], frame.Pix[off+2], frame.Pix[off+3] = b, g, r, 255
}
}
return frame
}
// ptr returns a pointer to v, for passing headers to WritePacket.
func ptr[T any](v T) *T { return &v }
// addr formats a port as a listen address.
func addr(port int) string {
return ":" + strconv.Itoa(port)
}
// testPattern renders a standard SMPTE colour bar with a moving white box at
// the given frame index.
func testPattern(w, h, frame int) *image.YCbCr {
img := image.NewYCbCr(image.Rect(0, 0, w, h), image.YCbCrSubsampleRatio420)
// 7 vertical colour bars (grey, yellow, cyan, green, magenta, red, blue).
bars := []color.RGBA{
{R: 191, G: 191, B: 191}, // 75% grey
{R: 191, G: 191, B: 0}, // yellow
{R: 0, G: 191, B: 191}, // cyan
{R: 0, G: 191, B: 0}, // green
{R: 191, G: 0, B: 191}, // magenta
{R: 191, G: 0, B: 0}, // red
{R: 0, G: 0, B: 191}, // blue
}
const barCount = 7
barW := w / barCount
const boxSize = 80
// Moving white box sweeps left→right across the lower black block.
boxMinX := (frame*(w+boxSize)/120)%(w+boxSize) - boxSize/2
buf := make([]color.RGBA, w*h)
for by := 0; by < h; by++ {
rowIsBars := by < h*2/3
for bx := 0; bx < w; bx++ {
var c color.RGBA
switch {
case rowIsBars:
idx := bx / barW
if idx >= barCount {
idx = barCount - 1
}
c = bars[idx]
case by%8 < 4 && bx > w/3 && bx < w*2/3:
c = color.RGBA{R: 255, G: 255, B: 255, A: 255}
default:
c = color.RGBA{}
}
if bx >= boxMinX && bx < boxMinX+boxSize && by >= h*2/3 {
c = color.RGBA{R: 255, G: 255, B: 255, A: 255}
}
buf[by*w+bx] = c
}
}
// Chroma planes: average each 2x2 RGB block, then convert to Cb/Cr.
for by := 0; by < h; by += 2 {
for bx := 0; bx < w; bx += 2 {
var rSum, gSum, bSum uint32
n := uint32(0)
for dy := 0; dy < 2; dy++ {
for dx := 0; dx < 2; dx++ {
xx, yy := bx+dx, by+dy
if xx >= w || yy >= h {
continue
}
px := buf[yy*w+xx]
rSum += uint32(px.R)
gSum += uint32(px.G)
bSum += uint32(px.B)
n++
}
}
_, cb, cr := color.RGBToYCbCr(uint8(rSum/n), uint8(gSum/n), uint8(bSum/n))
img.Cb[(by/2)*img.CStride+bx/2] = cb
img.Cr[(by/2)*img.CStride+bx/2] = cr
}
}
// Luma plane: Y = YCbCr luma of every pixel.
for by := 0; by < h; by++ {
for bx := 0; bx < w; bx++ {
px := buf[by*w+bx]
y, _, _ := color.RGBToYCbCr(px.R, px.G, px.B)
img.Y[by*img.YStride+bx] = y
}
}
return img
}
+92
View File
@@ -0,0 +1,92 @@
package flinger
import (
"net"
"testing"
"time"
)
// TestEnginePatternStartStop runs the engine with the synthetic pattern source
// and verifies it produces frames on the wire and stops cleanly.
func TestEnginePatternStartStop(t *testing.T) {
cfg := DefaultConfig()
cfg.Source = "pattern"
cfg.Port = 0 // ephemeral
eng, err := New(cfg)
if err != nil {
t.Fatalf("New: %v", err)
}
eng.Start()
defer eng.Stop()
// 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()))
if err != nil {
t.Fatalf("dial: %v", err)
}
defer func() { _ = conn.Close() }()
// The video loop sends ~30fps; wait for the counter to advance.
deadline := time.Now().Add(3 * time.Second)
_ = conn.SetReadDeadline(deadline)
// Drain whatever arrives while waiting for the frame counter to move.
go func() {
buf := make([]byte, 64*1024)
for {
if _, err := conn.Read(buf); err != nil {
return
}
}
}()
for time.Now().Before(deadline) {
if eng.Status().Frames > 0 {
break
}
time.Sleep(20 * time.Millisecond)
}
if eng.Status().Frames == 0 {
t.Error("expected frames to be counted")
}
}
// TestEngineConfigDefaults verifies DefaultConfig is sane.
func TestEngineConfigDefaults(t *testing.T) {
c := DefaultConfig()
if c.Port != 9756 {
t.Errorf("default port = %d, want 9756", c.Port)
}
if c.Source != "screen" {
t.Errorf("default source = %q, want screen", c.Source)
}
if c.Quality < 1 || c.Quality > 100 {
t.Errorf("default quality %d out of range", c.Quality)
}
}
// TestNewRejectsBadSource ensures invalid sources fail fast.
func TestNewRejectsBadSource(t *testing.T) {
cfg := DefaultConfig()
cfg.Source = "bogus"
if _, err := New(cfg); err == nil {
t.Error("expected error for unknown source")
}
}
func itoa(v int) string {
if v == 0 {
return "0"
}
var buf [20]byte
i := len(buf)
for v > 0 {
i--
buf[i] = byte('0' + v%10)
v /= 10
}
return string(buf[i:])
}