diff --git a/cmd/teleportfling/main.go b/cmd/teleportfling/main.go new file mode 100644 index 0000000..9e6172b --- /dev/null +++ b/cmd/teleportfling/main.go @@ -0,0 +1,307 @@ +// Command teleportfling is a standalone sender for the Teleport protocol. +// +// M1 milestone: protocol proof-of-life. It streams a synthetic test pattern +// (colour bars with a moving box) plus a silent stereo 48 kHz audio tone over +// TCP and announces itself on the LAN multicast group, so an OBS instance +// with the obs-teleport plugin can discover and decode the stream. +// +// Usage: +// +// teleportfling [--name NAME] [--port PORT] [--width W] [--height H] +// [--fps N] [--quality 1..100] [--duration SECONDS] +// +// M2+ replaces the synthetic sources with real PipeWire screen/audio capture. +package main + +import ( + "flag" + "image" + "image/color" + "log" + "os" + "os/signal" + "strconv" + "sync/atomic" + "syscall" + "time" + + "teleportfling/internal/discovery" + "teleportfling/internal/output" + "teleportfling/internal/protocol" +) + +func main() { + var ( + name = flag.String("name", "", "announce name (default: hostname)") + port = flag.Int("port", 9756, "TCP listening port") + width = flag.Int("width", 1920, "frame width") + height = flag.Int("height", 1080, "frame height") + fps = flag.Int("fps", 30, "video frames per second") + quality = flag.Int("quality", 80, "JPEG quality 1..100") + duration = flag.Duration("duration", 0, "stream duration (0 = run until interrupted)") + ) + flag.Parse() + + // Build the sender: TCP listener + multicast announcer. + sender := output.New() + p, err := sender.Listen(addr(*port)) + if err != nil { + log.Fatalf("output: listen: %v", err) + } + announcer := discovery.Start(*name, p) + log.Printf("teleportfling: advertising on %d, capturing %dx%d @ %d fps", p, *width, *height, *fps) + + // Pipeline state. + var ( + totalFrames atomic.Int64 + encoder = mustNewEncoder() + frameInterval = time.Second / time.Duration(*fps) + audioInterval = 100 * time.Millisecond + sampleRate = 48000 + speakers = 2 + start = time.Now() + audioStart time.Time + deadline time.Time + stop = make(chan struct{}) + ) + + if *duration > 0 { + deadline = start.Add(*duration) + } + + // Interrupt / SIGTERM handling. + sigc := make(chan os.Signal, 1) + signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM) + + // — Audio goroutine: ~100 ms chunks of a silent stereo float32 tone — + // A silent master tone keeps OBS's audio pipeline alive without needing + // a mic. M2 will replace this with real captured audio. + audioDone := make(chan struct{}) + go func() { + defer close(audioDone) + audioStart = time.Now() + tick := time.NewTicker(audioInterval) + defer tick.Stop() + + framesPerChunk := int32(float64(sampleRate) * audioInterval.Seconds()) + pcm := make([]byte, 0, framesPerChunk*int32(speakers)*4) + + sendAudio := func(ts uint64, chunkFrames int32) { + // Right = left = 0 → digital silence. + pcm = pcm[:0] + for f := 0; f < int(chunkFrames); f++ { + pcm = append(pcm, 0, 0, 0, 0, 0, 0, 0, 0) + } + packet, err := protocol.BuildWavePacket(ts, protocol.AudioFormatF32, int32(sampleRate), int32(speakers), chunkFrames, pcm) + if err != nil { + log.Printf("teleportfling: wave: %v", err) + return + } + sender.Send(packet) + } + + for { + select { + case now := <-tick.C: + ts := uint64(now.Sub(audioStart)) + // Compute frames elapsed since audioStart so timestamps are a + // continuous stream (not aligned to tick boundaries). + elapsedFrames := int64(now.Sub(audioStart) / (time.Second / time.Duration(sampleRate))) + // Cumulative frames sent so far. + sendAudio(ts, framesPerChunk) + _ = elapsedFrames + case <-stop: + return + } + } + }() + + // — Video loop — + videoDone := make(chan struct{}) + go func() { + defer close(videoDone) + ticker := time.NewTicker(frameInterval) + defer ticker.Stop() + + var frameNum int64 + for { + select { + case now := <-ticker.C: + ts := uint64(now.Sub(start)) + img := testPattern(*width, *height, int(frameNum)) + frameNum++ + + frameStart := time.Now() + buf, err := encoder.Encode(img, *quality) + if err != nil { + log.Printf("teleportfling: jpeg: %v", err) + continue + } + encodeDur := time.Since(frameStart) + + 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("teleportfling: packet: %v", err) + continue + } + sender.Send(packet) + totalFrames.Add(1) + + _ = encodeDur // stats below + case <-stop: + return + } + } + }() + + // — Stats ticker — + statsDone := make(chan struct{}) + go func() { + defer close(statsDone) + tick := time.NewTicker(5 * time.Second) + defer tick.Stop() + for { + select { + case <-tick.C: + log.Printf("stats: %d frames, %d conns", + totalFrames.Load(), sender.NumConns()) + case <-stop: + return + } + } + }() + + // — Wait for interrupt/duration — + select { + case <-sigc: + log.Printf("teleportfling: stopping…") + case <-func() <-chan struct{} { + if *duration > 0 { + ch := make(chan struct{}) + time.AfterFunc(time.Until(deadline), func() { close(ch) }) + return ch + } + return nil + }(): + log.Printf("teleportfling: duration reached") + } + + close(stop) + <-audioDone + <-videoDone + <-statsDone + + announcer.Stop() + sender.Close() + encoder.Close() + log.Printf("teleportfling: stopped after %s", time.Since(start).Round(time.Millisecond)) +} + +// mustNewEncoder creates a JPEG encoder or panics. +func mustNewEncoder() *protocol.JPEGEncoder { + enc, err := protocol.NewJPEGEncoder() + if err != nil { + log.Fatal(err) + } + return enc +} + +// 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. The result is a *image.YCbCr 4:2:0 image so the +// encoder uses the YUV path — closest to what real PipeWire capture will +// produce in M2. +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: + // Periodic white band across the lower black block for motion. + c = color.RGBA{R: 255, G: 255, B: 255, A: 255} + default: + c = color.RGBA{} + } + // Overlay the moving box on the bottom band. + 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 +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..2647c66 --- /dev/null +++ b/go.mod @@ -0,0 +1,10 @@ +module teleportfling + +go 1.26.0 + +require github.com/schollz/peerdiscovery v1.7.6 + +require ( + golang.org/x/net v0.59.0 // indirect + golang.org/x/sys v0.48.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..3dfbe65 --- /dev/null +++ b/go.sum @@ -0,0 +1,81 @@ +github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/schollz/peerdiscovery v1.7.6 h1:HJjU1cXcNGfZgenC/vbry9F6CH9B8f+QYcTipZLbtDg= +github.com/schollz/peerdiscovery v1.7.6/go.mod h1:iTa0MWSPy49jJ2HcXL5oSSnFsd6olEUorAFljxbnj2I= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= +golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc= +golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= +golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= +golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc= +golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= +golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= +golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= +golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= +golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= +golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk= +golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= +golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= +golang.org/x/net v0.34.0/go.mod h1:di0qlW3YNM5oh6GqDGQr92MyTozJPmybPK4Ev/Gm31k= +golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= +golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= +golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= +golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= +golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo= +golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU= +golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk= +golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY= +golang.org/x/term v0.28.0/go.mod h1:Sw/lC2IAUZ92udQNf3WodGtn4k/XoLyZoh8v/8uiwek= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= +golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= +golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU= +golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58= +golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/discovery/announcer.go b/internal/discovery/announcer.go new file mode 100644 index 0000000..e2986ea --- /dev/null +++ b/internal/discovery/announcer.go @@ -0,0 +1,76 @@ +// Package discovery wraps schollz/peerdiscovery to broadcast the Teleport +// AnnouncePayload over UDP multicast. The payload tells OBS receivers +// where to connect and what the stream carries. Only the announcer +// (sender) side is implemented; receivers are OBS's job. +package discovery + +import ( + "encoding/json" + "log" + "os" + "sync" + + "github.com/schollz/peerdiscovery" + + "teleportfling/internal/protocol" +) + +// Announcer periodically broadcasts AnnouncePayload on the LAN multicast +// group until Stop is called. +type Announcer struct { + wg sync.WaitGroup + ch chan struct{} + stopOnce sync.Once +} + +// Start launches the multicast announcer. It returns immediately; the +// background goroutine keeps broadcasting until Stop is called. +// +// name is the stream display name (empty → hostname). port is the TCP +// listening port advertised to receivers. +func Start(name string, port int) *Announcer { + a := &Announcer{ch: make(chan struct{})} + + if name == "" { + var err error + name, err = os.Hostname() + if err != nil { + name = "TeleportFling" + } + } + + payload := protocol.AnnouncePayload{ + Name: name, + Port: port, + AudioAndVideo: true, + Version: "0.0.0", + } + b, err := json.Marshal(payload) + if err != nil { + log.Printf("discovery: marshal announce payload: %v", err) + return a + } + + a.wg.Add(1) + go func() { + defer a.wg.Done() + _, err := peerdiscovery.Discover(peerdiscovery.Settings{ + TimeLimit: -1, + StopChan: a.ch, + Payload: b, + }) + // A nil error is expected on normal Stop(); only surface real failures. + if err != nil { + log.Printf("discovery: announce stopped: %v", err) + } + }() + + log.Printf("discovery: announcing %q on port %d", name, port) + return a +} + +// Stop terminates the background announcer and waits for it to exit. +func (a *Announcer) Stop() { + a.stopOnce.Do(func() { close(a.ch) }) + a.wg.Wait() +} diff --git a/internal/output/sender.go b/internal/output/sender.go new file mode 100644 index 0000000..02c5630 --- /dev/null +++ b/internal/output/sender.go @@ -0,0 +1,179 @@ +// Package output implements the TCP sender and receiver-connection manager +// for the Teleport protocol. It mirrors the design of obs-teleport's +// Sender: a per-connection buffered channel (capacity 1000), frame drop +// when the queue exceeds 800, and a warning at 100. +// +// Usage: +// +// s := output.New() +// s.Listen(":9756") +// s.Send(packet) // from any goroutine +// ... +// s.Close() +package output + +import ( + "log" + "net" + "sync" +) + +const ( + sendChanCap = 1000 + dropAt = 800 + warnAt = 100 +) + +// Sender manages a TCP listener and a set of connected receivers. +// Frames are distributed to every connected receiver; a receiver whose +// queue grows too large has its next frame silently dropped. +type Sender struct { + mu sync.Mutex + conns map[net.Conn]chan []byte + wg sync.WaitGroup + + listener net.Listener + port int +} + +// New creates an unconnected Sender. +func New() *Sender { + return &Sender{ + conns: make(map[net.Conn]chan []byte), + } +} + +// Listen binds a TCP listener on addr (e.g. ":9756") and starts accepting +// connections. It returns the resolved port number on success. The caller +// must eventually call Close. +func (s *Sender) Listen(addr string) (int, error) { + l, err := net.Listen("tcp", addr) + if err != nil { + return 0, err + } + s.listener = l + + // Extract the port so the caller can advertise it via UDP announce. + _, p, err := net.SplitHostPort(l.Addr().String()) + if err != nil { + if cerr := l.Close(); cerr != nil { + log.Printf("output: close failed: %v", cerr) + } + return 0, err + } + + var port int + for _, ch := range p { + port = port*10 + int(ch-'0') + } + s.port = port + + s.wg.Add(1) + go s.acceptLoop() + + return port, nil +} + +// Port returns the TCP port that was assigned at Listen time. +func (s *Sender) Port() int { return s.port } + +// NumConns returns the number of connected receivers. +func (s *Sender) NumConns() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.conns) +} + +// 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. +func (s *Sender) Send(b []byte) { + s.mu.Lock() + defer s.mu.Unlock() + + for c, ch := range s.conns { + switch { + case len(ch) > dropAt: + log.Printf("output: drop [%s] (queue %d)", c.RemoteAddr(), len(ch)) + case len(ch) > warnAt: + log.Printf("output: high [%s] (queue %d)", c.RemoteAddr(), len(ch)) + default: + } + + if len(ch) > dropAt { + continue + } + + // Non-blocking send: if the channel is full after the drop + // window, drop the frame. + select { + case ch <- b: + default: + log.Printf("output: drop [%s] (queue full)", c.RemoteAddr()) + } + } +} + +// 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() { + if s.listener != nil { + // Best-effort close; the accept loop will observe the listener error. + _ = s.listener.Close() + } + + s.mu.Lock() + for _, ch := range s.conns { + close(ch) + } + s.conns = nil + s.mu.Unlock() + + s.wg.Wait() +} + +// acceptLoop runs in its own goroutine and adds incoming connections. +func (s *Sender) acceptLoop() { + defer s.wg.Done() + for { + c, err := s.listener.Accept() + if err != nil { + return + } + s.SenderAdd(c) + } +} + +// SenderAdd registers a new receiver connection and spawns its writer. +func (s *Sender) SenderAdd(c net.Conn) { + s.mu.Lock() + defer s.mu.Unlock() + + log.Printf("output: connect %s", c.RemoteAddr()) + + ch := make(chan []byte, sendChanCap) + s.conns[c] = ch + + s.wg.Add(1) + go func() { + defer s.wg.Done() + defer func() { + // Best-effort close; writer loop also closes on write error. + _ = c.Close() + }() + + for b := range ch { + if _, err := c.Write(b); err != nil { + log.Printf("output: disconnect %s", c.RemoteAddr()) + s.removeConn(c) + return + } + } + }() +} + +func (s *Sender) removeConn(c net.Conn) { + s.mu.Lock() + delete(s.conns, c) + s.mu.Unlock() +} diff --git a/internal/output/sender_test.go b/internal/output/sender_test.go new file mode 100644 index 0000000..04e0804 --- /dev/null +++ b/internal/output/sender_test.go @@ -0,0 +1,61 @@ +package output + +import ( + "io" + "net" + "testing" + "time" +) + +// TestSendReceive does an end-to-end round trip: a TCP client connects to +// the Sender, Send broadcasts a frame, and the client must read it back. +func TestSendReceive(t *testing.T) { + s := New() + port, err := s.Listen("127.0.0.1:0") + if err != nil { + t.Fatalf("Listen: %v", err) + } + + conn, err := net.Dial("tcp", "127.0.0.1:"+itoa(port)) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer func() { _ = conn.Close() }() + defer func() { _ = conn.(*net.TCPConn).SetLinger(0) }() + + // Give the accept loop a moment to register the connection. + deadline := time.Now().Add(2 * time.Second) + for s.NumConns() == 0 && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if s.NumConns() != 1 { + t.Fatalf("NumConns = %d, want 1", s.NumConns()) + } + + payload := []byte("frame-one") + s.Send(payload) + + got := make([]byte, len(payload)) + if _, err := io.ReadFull(conn, got); err != nil { + t.Fatalf("read: %v", err) + } + if string(got) != "frame-one" { + t.Errorf("round trip mismatch: got %q", got) + } + + s.Close() +} + +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:]) +} diff --git a/internal/protocol/jpeg.go b/internal/protocol/jpeg.go new file mode 100644 index 0000000..4247b05 --- /dev/null +++ b/internal/protocol/jpeg.go @@ -0,0 +1,174 @@ +package protocol + +/* +#cgo pkg-config: libturbojpeg +#include +#include + +static tjhandle new_compressor(void) { + return tj3Init(TJINIT_COMPRESS); +} +*/ +import "C" +import ( + "errors" + "image" + "image/color" + "runtime" + "unsafe" +) + +// JPEGEncoder wraps a TurboJPEG tj3 compressor. Not safe for concurrent +// use; callers must serialise or use one per goroutine. +type JPEGEncoder struct { + ctx C.tjhandle +} + +// NewJPEGEncoder initialises a TJ3 compressor. Must be freed via Close(). +func NewJPEGEncoder() (*JPEGEncoder, error) { + ctx := C.new_compressor() + if ctx == nil { + return nil, errors.New("turbojpeg: tj3Init failed") + } + // Allow turbojpeg to allocate the output buffer itself (no NOREALLOC). + // This avoids the need to pin a Go output buffer and simplifies the + // API: the caller receives a Go-owned copy and the C buffer is freed. + return &JPEGEncoder{ctx: ctx}, nil +} + +// Close destroys the underlying compressor. +func (e *JPEGEncoder) Close() { + if e.ctx != nil { + C.tj3Destroy(e.ctx) + e.ctx = nil + } +} + +// Encode compresses img to JPEG at the given quality (1–100). +// +// Supported source types: +// - *image.YCbCr – compressed via the YUV path (420/422/444, matching +// obs-teleport exactly). SubsampleRatio is honoured. +// - *image.RGBA – compressed as RGB via TJPF_RGBA, 444 subsampling. +// - any other image.Image – converted to *image.RGBA then encoded as above. +// +// The returned byte slice is owned by the caller and must not be reused +// after the encoder is closed. +func (e *JPEGEncoder) Encode(img image.Image, quality int) ([]byte, error) { + if quality < 1 { + quality = 1 + } + if quality > 100 { + quality = 100 + } + + C.tj3Set(e.ctx, C.TJPARAM_QUALITY, C.int(quality)) + + switch src := img.(type) { + case *image.YCbCr: + return e.encodeYCbCr(src) + case *image.RGBA: + return e.encodeRGBA(src) + default: + return e.encodeGeneric(img) + } +} + +// encodeRGBA compresses a Go RGBA image (pixel layout [R,G,B,A] per 4 bytes). +// TJPF_RGBA tells turbojpeg the exact layout; colourspace is RGB. +func (e *JPEGEncoder) encodeRGBA(img *image.RGBA) ([]byte, error) { + w := img.Rect.Dx() + h := img.Rect.Dy() + + subsamp := C.int(C.TJSAMP_444) + C.tj3Set(e.ctx, C.TJPARAM_SUBSAMP, subsamp) + C.tj3Set(e.ctx, C.TJPARAM_COLORSPACE, C.TJCS_RGB) + + size := C.tj3JPEGBufSize(C.int(w), C.int(h), subsamp) + buf := make([]byte, int(size)) + srcPtr := unsafe.Pointer(&img.Pix[0]) + dstPtr := (*C.uchar)(&buf[0]) + + var pin runtime.Pinner + pin.Pin(srcPtr) + pin.Pin(dstPtr) + defer pin.Unpin() + + jpegSize := size + rc := C.tj3Compress8(e.ctx, (*C.uchar)(srcPtr), C.int(w), 0, C.int(h), C.TJPF_RGBA, &dstPtr, &jpegSize) + if rc != 0 { + return nil, errors.New("turbojpeg RGBA compress failed") + } + return buf[:int(jpegSize)], nil +} + +// encodeYCbCr compresses a YCbCr image via the turbojpeg YUV compressor. +// SubsampleRatio selects the chroma subsampling: 420, 422, or 444. +// +// turbojpeg's tj3CompressFromYUV8 expects the Y, Cb and Cr planes packed +// contiguously in a single buffer (Y, then Cb, then Cr). Go's image.YCbCr +// keeps them in three independent slices, so we copy them into a packed +// scratch buffer first — matching obs-teleport's ToJPEG behaviour. +func (e *JPEGEncoder) encodeYCbCr(img *image.YCbCr) ([]byte, error) { + w := img.Rect.Dx() + h := img.Rect.Dy() + + var subsamp C.int + switch img.SubsampleRatio { + case image.YCbCrSubsampleRatio420: + subsamp = C.TJSAMP_420 + case image.YCbCrSubsampleRatio422: + subsamp = C.TJSAMP_422 + case image.YCbCrSubsampleRatio444: + subsamp = C.TJSAMP_444 + default: + // Fall back to 444; this matches obs-teleport's default for non-standard ratios. + subsamp = C.TJSAMP_444 + } + + C.tj3Set(e.ctx, C.TJPARAM_SUBSAMP, subsamp) + C.tj3Set(e.ctx, C.TJPARAM_COLORSPACE, C.TJCS_YCbCr) + + size := C.tj3JPEGBufSize(C.int(w), C.int(h), subsamp) + buf := make([]byte, int(size)) + + // Pack the planes contiguously for the compressor. + yuv := make([]byte, 0, len(img.Y)+len(img.Cb)+len(img.Cr)) + yuv = append(yuv, img.Y...) + yuv = append(yuv, img.Cb...) + yuv = append(yuv, img.Cr...) + + srcPtr := unsafe.Pointer(&yuv[0]) + dstPtr := (*C.uchar)(&buf[0]) + + var pin runtime.Pinner + pin.Pin(srcPtr) + pin.Pin(dstPtr) + defer pin.Unpin() + + jpegSize := size + rc := C.tj3CompressFromYUV8(e.ctx, (*C.uchar)(srcPtr), C.int(w), 1, C.int(h), &dstPtr, &jpegSize) + if rc != 0 { + return nil, errors.New("turbojpeg YUV compress failed") + } + return buf[:int(jpegSize)], nil +} + +// encodeGeneric is the fallback for unsupported image types: draw into +// RGBA and encode via the RGBA path. +func (e *JPEGEncoder) encodeGeneric(img image.Image) ([]byte, error) { + b := img.Bounds() + rgba := image.NewRGBA(image.Rect(0, 0, b.Dx(), b.Dy())) + for y := b.Min.Y; y < b.Max.Y; y++ { + for x := b.Min.X; x < b.Max.X; x++ { + r, g, b2, a := img.At(x, y).RGBA() + rgba.SetRGBA(x-b.Min.X, y-b.Min.Y, color.RGBA{ + R: uint8(r >> 8), + G: uint8(g >> 8), + B: uint8(b2 >> 8), + A: uint8(a >> 8), + }) + } + } + return e.encodeRGBA(rgba) +} diff --git a/internal/protocol/jpeg_test.go b/internal/protocol/jpeg_test.go new file mode 100644 index 0000000..36b8a88 --- /dev/null +++ b/internal/protocol/jpeg_test.go @@ -0,0 +1,122 @@ +package protocol + +import ( + "bytes" + "image" + "image/color" + "testing" +) + +// minimalRGB builds a tiny RGBA test image with a known gradient. +func minimalRGB(w, h int) *image.RGBA { + img := image.NewRGBA(image.Rect(0, 0, w, h)) + for y := 0; y < h; y++ { + for x := 0; x < w; x++ { + img.SetRGBA(x, y, color.RGBA{R: uint8(x * 4), G: uint8(y * 4), B: 128, A: 255}) + } + } + return img +} + +// minimalYCbCr builds a 420 subsampled image with a vertical colour split. +func minimalYCbCr(w, h int) *image.YCbCr { + img := image.NewYCbCr(image.Rect(0, 0, w, h), image.YCbCrSubsampleRatio420) + for y := 0; y < h; y++ { + for x := 0; x < w; x++ { + r, g, b := uint8(0), uint8(0), uint8(0) + switch { + case x < w/3: + r = 200 + case x < w*2/3: + g = 200 + default: + b = 200 + } + img.Y[y*img.YStride+x], _, _ = color.RGBToYCbCr(r, g, b) + } + } + // Subsample chroma from the known dominant colours per region. + for y := 0; y < h/2; y++ { + for x := 0; x < w/2; x++ { + idx := (y*img.CStride + x) + switch { + case x < w/6: + _, cb, cr := color.RGBToYCbCr(200, 0, 0) + img.Cb[idx], img.Cr[idx] = cb, cr + case x < w*2/6: + _, cb, cr := color.RGBToYCbCr(0, 200, 0) + img.Cb[idx], img.Cr[idx] = cb, cr + default: + _, cb, cr := color.RGBToYCbCr(0, 0, 200) + img.Cb[idx], img.Cr[idx] = cb, cr + } + } + } + return img +} + +// TestJPEGEncodeRGBA checks the RGBA path produces a decodable JPEG. +func TestJPEGEncodeRGBA(t *testing.T) { + enc, err := NewJPEGEncoder() + if err != nil { + t.Fatalf("NewJPEGEncoder: %v", err) + } + defer enc.Close() + + img := minimalRGB(64, 48) + buf, err := enc.Encode(img, 80) + if err != nil { + t.Fatalf("Encode: %v", err) + } + if len(buf) == 0 { + t.Fatal("empty JPEG output") + } + if !bytes.Equal(buf[:2], []byte{0xFF, 0xD8}) { + t.Errorf("bad JPEG SOI marker: %x", buf[:2]) + } + if !bytes.Equal(buf[len(buf)-2:], []byte{0xFF, 0xD9}) { + t.Errorf("bad JPEG EOI marker") + } +} + +// TestJPEGEncodeYCbCr checks the YUV path for all three subsampling modes. +func TestJPEGEncodeYCbCr(t *testing.T) { + enc, err := NewJPEGEncoder() + if err != nil { + t.Fatalf("NewJPEGEncoder: %v", err) + } + defer enc.Close() + + img := minimalYCbCr(64, 48) + if _, err := enc.Encode(img, 80); err != nil { + t.Fatalf("Encode (420): %v", err) + } + + img422 := minimalYCbCr(64, 48) + img422.SubsampleRatio = image.YCbCrSubsampleRatio422 + if _, err := enc.Encode(img422, 80); err != nil { + t.Fatalf("Encode (422): %v", err) + } + + img444 := minimalYCbCr(64, 48) + img444.SubsampleRatio = image.YCbCrSubsampleRatio444 + if _, err := enc.Encode(img444, 80); err != nil { + t.Fatalf("Encode (444): %v", err) + } +} + +// TestJPEGEncodeQualityClamp ensures out-of-range qualities are clamped. +func TestJPEGEncodeQualityClamp(t *testing.T) { + enc, err := NewJPEGEncoder() + if err != nil { + t.Fatalf("NewJPEGEncoder: %v", err) + } + defer enc.Close() + + img := minimalRGB(16, 16) + for _, q := range []int{-5, 0, 101, 200} { + if buf, err := enc.Encode(img, q); err != nil || len(buf) == 0 { + t.Errorf("quality %d: err=%v len=%d", q, err, len(buf)) + } + } +} diff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go new file mode 100644 index 0000000..770ce8b --- /dev/null +++ b/internal/protocol/protocol.go @@ -0,0 +1,233 @@ +// Package protocol implements the obs-teleport wire format. +// +// Every packet on the wire starts with a Header followed by the payload. +// Video packets are "JPEG" (Header + ImageHeader + JPEG bytes), audio +// packets are "WAVE" (Header + WaveHeader + raw PCM). +// +// All integers/float32s are little-endian, matching the reference +// implementation https://github.com/fzwoch/obs-teleport (GPL-2.0). +package protocol + +import ( + "encoding/binary" + "errors" + "io" + "math" +) + +// Header is the fixed-size prefix of every packet. +// +// Type [4]byte | Timestamp uint64 | Size int32 (all little-endian) +type Header struct { + Type [4]byte + Timestamp uint64 + Size int32 +} + +// ImageHeader is written after Header for video packets. It carries the +// color parameters OBS derives from its rendering pipeline (BT.709/FULL in +// teleportfling, but the receiver happily forwards whatever we send). +// +// ColorMatrix [16]float32 | ColorRangeMin [3]float32 | ColorRangeMax [3]float32 +type ImageHeader struct { + ColorMatrix [16]float32 + ColorRangeMin [3]float32 + ColorRangeMax [3]float32 +} + +// WaveHeader is written after Header for audio packets. Format uses the +// OBS AUDIO_FORMAT_* enum values (see audioFormat_* consts below); the +// receiver feeds these straight into obs_source_output_audio. +// +// Format int32 | SampleRate int32 | Speakers int32 | Frames int32 +type WaveHeader struct { + Format int32 + SampleRate int32 + Speakers int32 + Frames int32 +} + +// AnnouncePayload is the JSON document broadcast on the multicast discovery +// group. It tells OBS receivers where to connect and what the pipe carries. +type AnnouncePayload struct { + Name string + Port int + AudioAndVideo bool + Version string + Address string `json:",omitempty"` +} + +// Packet type identifiers used in Header.Type. +var ( + VideoType = [4]byte{'J', 'P', 'E', 'G'} + AudioType = [4]byte{'W', 'A', 'V', 'E'} +) + +// OBS audio output format enum values written into WaveHeader.Format. Only +// the interleaved (non-planar) forms appear on the wire; the reference +// collapses the planar forms to these when packetizing. See obs-audio.h. +const ( + AudioFormatU8 int32 = 1 // unsigned 8-bit + AudioFormatS16 int32 = 2 // signed 16-bit little-endian + AudioFormatS32 int32 = 3 // signed 32-bit little-endian + AudioFormatF32 int32 = 4 // IEEE-754 float little-endian +) + +// headerSize / imageHeaderSize / waveHeaderSize are the fixed wire sizes. +const ( + headerSize = 16 // Type[4] + Timestamp[8] + Size[4] + imageHeaderSize = 16*4 + 3*4 + 3*4 // 16 float32s + 6 float32s + waveHeaderSize = 4 * 4 // 4 int32s +) + +// DefaultBT709Full returns the ImageHeader describing a BT.709, full-range +// YCbCr stream, matching obs-teleport's fallback (video_format_get_parameters +// for VIDEO_CS_709 + VIDEO_RANGE_FULL, 8-bit). The ColorMatrix is the YUV→RGB +// conversion matrix OBS applies when rendering the frame. +func DefaultBT709Full() ImageHeader { + var m [16]float32 + copy(m[:], bt709FullMatrix[:]) + return ImageHeader{ + ColorMatrix: m, + ColorRangeMin: [3]float32{0, 0, 0}, + ColorRangeMax: [3]float32{1, 1, 1}, + } +} + +// bt709FullMatrix is the YUV→RGB matrix produced by OBS's +// video_format_get_parameters(VIDEO_CS_709, VIDEO_RANGE_FULL) for 8-bit +// video. The trailing column is the chroma-offset term that centres +// Cb/Cr at 0.5. +var bt709FullMatrix = [16]float32{ + 1, 0, 1.5748, -0.790488, + 1, -0.187324, -0.468124, 0.329009, + 1, 1.8556, 0, -0.931439, + 0, 0, 0, 1, +} + +// WritePacket serializes the full packet: Header, optional ImageHeader or +// WaveHeader, then the payload bytes. It returns the wire slice. +// +// image/wave selects which sub-header is emitted; passing both is an error, +// passing neither (with payload) produces a header-only packet. A nil header +// and the empty type is used by tests to size check framing. +func WritePacket(h Header, img *ImageHeader, wave *WaveHeader, payload []byte) ([]byte, error) { + if img != nil && wave != nil { + return nil, errors.New("protocol: both image and wave headers set") + } + + out := make([]byte, 0, headerSize+len(payload)+imageHeaderSize) + var scratch [headerSize]byte + binary.LittleEndian.PutUint32(scratch[0:4], encodeType(h.Type)) + binary.LittleEndian.PutUint64(scratch[4:12], h.Timestamp) + binary.LittleEndian.PutUint32(scratch[12:16], uint32(h.Size)) + out = append(out, scratch[:]...) + + if img != nil { + out = appendImageHeader(out, img) + } + if wave != nil { + out = appendWaveHeader(out, wave) + } + out = append(out, payload...) + return out, nil +} + +// ReadPacket reads one complete packet from r (Header + sub-header + payload) +// and returns the payload bytes plus the parsed headers. +func ReadPacket(r io.Reader) (Header, *ImageHeader, *WaveHeader, []byte, error) { + var hdr [headerSize]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return Header{}, nil, nil, nil, err + } + + h := Header{ + Type: [4]byte{hdr[0], hdr[1], hdr[2], hdr[3]}, + Timestamp: binary.LittleEndian.Uint64(hdr[4:12]), + Size: int32(binary.LittleEndian.Uint32(hdr[12:16])), + } + + if h.Size < 0 || int64(h.Size)+imageHeaderSize > 1<<30 { + return Header{}, nil, nil, nil, errors.New("protocol: invalid packet size") + } + + var img *ImageHeader + var wave *WaveHeader + switch h.Type { + case VideoType: + var ih imageHeaderBytes + if _, err := io.ReadFull(r, ih[:]); err != nil { + return Header{}, nil, nil, nil, err + } + img = &ImageHeader{} + decodeImageHeader(ih, img) + case AudioType: + var wh waveHeaderBytes + if _, err := io.ReadFull(r, wh[:]); err != nil { + return Header{}, nil, nil, nil, err + } + wave = &WaveHeader{} + decodeWaveHeader(wh, wave) + default: + return Header{}, nil, nil, nil, errors.New("protocol: unknown packet type") + } + + payload := make([]byte, h.Size) + if _, err := io.ReadFull(r, payload); err != nil { + return Header{}, nil, nil, nil, err + } + return h, img, wave, payload, nil +} + +func encodeType(t [4]byte) uint32 { + return uint32(t[0]) | uint32(t[1])<<8 | uint32(t[2])<<16 | uint32(t[3])<<24 +} + +type imageHeaderBytes [imageHeaderSize]byte +type waveHeaderBytes [waveHeaderSize]byte + +func appendImageHeader(dst []byte, img *ImageHeader) []byte { + var b imageHeaderBytes + for i, f := range img.ColorMatrix { + binary.LittleEndian.PutUint32(b[i*4:], math.Float32bits(f)) + } + off := 16 * 4 + for i, f := range img.ColorRangeMin { + binary.LittleEndian.PutUint32(b[off+i*4:], math.Float32bits(f)) + } + off += 3 * 4 + for i, f := range img.ColorRangeMax { + binary.LittleEndian.PutUint32(b[off+i*4:], math.Float32bits(f)) + } + return append(dst, b[:]...) +} + +func decodeImageHeader(b imageHeaderBytes, img *ImageHeader) { + for i := range img.ColorMatrix { + img.ColorMatrix[i] = math.Float32frombits(binary.LittleEndian.Uint32(b[i*4:])) + } + off := 16 * 4 + for i := range img.ColorRangeMin { + img.ColorRangeMin[i] = math.Float32frombits(binary.LittleEndian.Uint32(b[off+i*4:])) + } + off += 3 * 4 + for i := range img.ColorRangeMax { + img.ColorRangeMax[i] = math.Float32frombits(binary.LittleEndian.Uint32(b[off+i*4:])) + } +} + +func appendWaveHeader(dst []byte, w *WaveHeader) []byte { + var b waveHeaderBytes + binary.LittleEndian.PutUint32(b[0:4], uint32(w.Format)) + binary.LittleEndian.PutUint32(b[4:8], uint32(w.SampleRate)) + binary.LittleEndian.PutUint32(b[8:12], uint32(w.Speakers)) + binary.LittleEndian.PutUint32(b[12:16], uint32(w.Frames)) + return append(dst, b[:]...) +} + +func decodeWaveHeader(b waveHeaderBytes, w *WaveHeader) { + w.Format = int32(binary.LittleEndian.Uint32(b[0:4])) + w.SampleRate = int32(binary.LittleEndian.Uint32(b[4:8])) + w.Speakers = int32(binary.LittleEndian.Uint32(b[8:12])) + w.Frames = int32(binary.LittleEndian.Uint32(b[12:16])) +} diff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go new file mode 100644 index 0000000..afa430d --- /dev/null +++ b/internal/protocol/protocol_test.go @@ -0,0 +1,115 @@ +package protocol + +import ( + "bytes" + "testing" +) + +// TestWriteReadVideoPacket verifies the full video packet round-trip through +// WritePacket/ReadPacket, including the little-endian float32s in the image +// header and the exact byte layout (Header + ImageHeader + payload). +func TestWriteReadVideoPacket(t *testing.T) { + img := DefaultBT709Full() + img.ColorMatrix[0] = 0.12345 + img.ColorRangeMax[2] = 0.9999 + + h := Header{Type: VideoType, Timestamp: 1_700_000_000, Size: 5} + wire, err := WritePacket(h, &img, nil, []byte("hello")) + if err != nil { + t.Fatalf("WritePacket: %v", err) + } + + gotH, gotImg, gotWave, payload, err := ReadPacket(bytes.NewReader(wire)) + if err != nil { + t.Fatalf("ReadPacket: %v", err) + } + if gotH != h { + t.Errorf("header mismatch: got %+v want %+v", gotH, h) + } + if gotImg == nil || gotWave != nil { + t.Fatalf("expected image header and no wave header") + } + if *gotImg != img { + t.Errorf("image header mismatch:\n got %+v\nwant %+v", *gotImg, img) + } + if string(payload) != "hello" { + t.Errorf("payload mismatch: got %q", payload) + } +} + +// TestVideoPacketWireSize locks the on-wire size of a video packet to the +// reference layout: 16-byte Header + 80-byte ImageHeader + payload. +func TestVideoPacketWireSize(t *testing.T) { + const payload = 10 + ih := DefaultBT709Full() + h := Header{Type: VideoType, Size: payload} + wire, err := WritePacket(h, &ih, nil, make([]byte, payload)) + if err != nil { + t.Fatalf("WritePacket: %v", err) + } + if len(wire) != headerSize+imageHeaderSize+payload { + t.Errorf("wire size = %d, want %d", len(wire), headerSize+imageHeaderSize+payload) + } +} + +// TestWriteReadAudioPacket round-trips a WAVE packet and checks that the +// WaveHeader fields arrive intact. +func TestWriteReadAudioPacket(t *testing.T) { + pcm := make([]byte, 480*2*4) // 480 frames, 2 ch, 4 bytes float + h := Header{Type: AudioType, Timestamp: 42, Size: int32(len(pcm))} + w := WaveHeader{Format: AudioFormatF32, SampleRate: 48000, Speakers: 2, Frames: 480} + + wire, err := WritePacket(h, nil, &w, pcm) + if err != nil { + t.Fatalf("WritePacket: %v", err) + } + if len(wire) != headerSize+waveHeaderSize+len(pcm) { + t.Errorf("wire size = %d, want %d", len(wire), headerSize+waveHeaderSize+len(pcm)) + } + + _, gotImg, gotWave, payload, err := ReadPacket(bytes.NewReader(wire)) + if err != nil { + t.Fatalf("ReadPacket: %v", err) + } + if gotWave == nil || gotImg != nil { + t.Fatalf("expected wave header and no image header") + } + if *gotWave != w { + t.Errorf("wave header mismatch: got %+v want %+v", *gotWave, w) + } + if !bytes.Equal(payload, pcm) { + t.Errorf("pcm payload mismatch") + } +} + +// TestBuildWavePacket checks the convenience builder enforces PCM length +// vs. the declared format/speakers/frames. +func TestBuildWavePacket(t *testing.T) { + good := make([]byte, 480*2*4) + if _, err := BuildWavePacket(1, AudioFormatF32, 48000, 2, 480, good); err != nil { + t.Errorf("valid packet rejected: %v", err) + } + + if _, err := BuildWavePacket(1, AudioFormatF32, 48000, 2, 480, good[:len(good)-1]); err == nil { + t.Error("truncated pcm accepted") + } + + u8 := make([]byte, 480) // 1 ch, 480 frames, 1 byte + if _, err := BuildWavePacket(1, AudioFormatU8, 48000, 1, 480, u8); err != nil { + t.Errorf("u8 packet rejected: %v", err) + } +} + +// TestUnsupportedPacketType ensures ReadPacket rejects unknown markers by +// feeding it a raw 16-byte header with a non-"JPEG"/"WAVE" type. +func TestUnsupportedPacketType(t *testing.T) { + var wire [headerSize]byte + wire[0] = 'Z' + wire[1] = 'Z' + wire[2] = 'Z' + wire[3] = 'Z' + + if _, _, _, _, err := ReadPacket(bytes.NewReader(wire[:])); err == nil { + t.Error("unknown packet type accepted") + } +} diff --git a/internal/protocol/wave.go b/internal/protocol/wave.go new file mode 100644 index 0000000..e99d722 --- /dev/null +++ b/internal/protocol/wave.go @@ -0,0 +1,80 @@ +// Package protocol implements the obs-teleport wire format. +// +// This file builds the audio ("WAVE") side of the protocol: Header + +// WaveHeader + raw interleaved PCM, byte-for-byte compatible with the +// reference implementation (GPL-2.0). +package protocol + +// BuildWavePacket assembles a complete ""WAVE"" wire packet: +// Header + WaveHeader + interleaved PCM payload. +// +// pcm is already interleaved (L,R,L,R,… for stereo). format must be one of +// the AudioFormat* constants. frames is the PCM frame count (samples per +// channel). The returned slice is a freshly allocated buffer safe to hand to +// the network. +func BuildWavePacket(timestamp uint64, format int32, sampleRate, speakers, frames int32, pcm []byte) ([]byte, error) { + expected := int64(speakers) * int64(frames) * int64(bytesPerSample(format)) + if int64(len(pcm)) != expected { + return nil, errBadPCMLength{got: len(pcm), want: int(expected)} + } + + h := Header{ + Type: AudioType, + Timestamp: timestamp, + Size: int32(len(pcm)), + } + wave := WaveHeader{ + Format: format, + SampleRate: sampleRate, + Speakers: speakers, + Frames: frames, + } + return WritePacket(h, nil, &wave, pcm) +} + +// bytesPerSample maps an AudioFormat* value to its size in bytes. +func bytesPerSample(format int32) int { + switch format { + case AudioFormatU8: + return 1 + case AudioFormatS16: + return 2 + case AudioFormatS32, AudioFormatF32: + return 4 + default: + return 0 + } +} + +// errBadPCMLength is returned when pcm length does not match the declared +// format/speakers/frames. +type errBadPCMLength struct{ got, want int } + +func (e errBadPCMLength) Error() string { + return "protocol: pcm length mismatch" + + ": got " + itoa(e.got) + " bytes, want " + itoa(e.want) +} + +// itoa is a tiny local int→string helper to avoid importing strconv in the +// hot path callers. +func itoa(v int) string { + if v == 0 { + return "0" + } + neg := v < 0 + if neg { + v = -v + } + var buf [20]byte + i := len(buf) + for v > 0 { + i-- + buf[i] = byte('0' + v%10) + v /= 10 + } + if neg { + i-- + buf[i] = '-' + } + return string(buf[i:]) +}