Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 22 additions & 1 deletion server/internal/capture/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt

createPipeline := func() (string, error) {
if pipelineConf.GstPipeline != "" {
if config.Wayland {
return "", errors.New("custom video pipelines are not supported with Wayland capture")
}
// replace {display} with valid display
return strings.Replace(pipelineConf.GstPipeline, "{display}", config.Display, 1), nil
}
Expand All @@ -49,6 +52,18 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
return "", err
}

if config.Wayland {
fps := screen.Rate
if fps <= 0 {
fps = 25
}
return fmt.Sprintf(
"appsrc name=appsrc is-live=true format=time do-timestamp=true "+
"caps=video/x-raw,format=BGRx,width=%d,height=%d,framerate=%d/1 "+
"%s ! appsink name=appsink", screen.Width, screen.Height, fps, pipeline,
), nil

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Appsrc framerate conflicts with pipeline

High Severity

Wayland appsrc caps and wf-recorder both use screen.Rate, while the encoding chain from GetPipeline often forces a different framerate via VideoConfig.Fps (default "25"). Fixed appsrc caps cannot renegotiate against that capsfilter, so the default desktop rate (30) yields a not-negotiated pipeline and live video never starts.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 0cd3a36. Configure here.

}

return fmt.Sprintf(
"ximagesrc display-name=%s show-pointer=%v use-damage=false "+
"%s ! appsink name=appsink", config.Display, pipelineConf.ShowPointer, pipeline,
Expand All @@ -69,7 +84,13 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
Msg("syntax check for video stream pipeline passed")

// append to videos
videos[video_id] = streamSinkNew(config.VideoCodec, createPipeline, video_id)
video := streamSinkNew(config.VideoCodec, createPipeline, video_id)
if config.Wayland {
video.SetFrameSourceFactory(func() (frameSource, error) {
return newWaylandFrameSource(config.WaylandRecorder, desktop.GetScreenSize()), nil
})
}
videos[video_id] = video
}

return &CaptureManagerCtx{
Expand Down
36 changes: 32 additions & 4 deletions server/internal/capture/streamsink.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,10 +32,12 @@ type StreamSinkManagerCtx struct {
mu sync.Mutex
wg sync.WaitGroup

codec codec.RTPCodec
pipeline gst.Pipeline
pipelineMu sync.Mutex
pipelineFn func() (string, error)
codec codec.RTPCodec
pipeline gst.Pipeline
pipelineMu sync.Mutex
pipelineFn func() (string, error)
frameSourceFn func() (frameSource, error)
frameSource frameSource

listeners map[uintptr]types.SampleListener
listenersKf map[uintptr]types.SampleListener // keyframe lobby
Expand Down Expand Up @@ -142,6 +144,10 @@ func (manager *StreamSinkManagerCtx) ID() string {
return manager.id
}

func (manager *StreamSinkManagerCtx) SetFrameSourceFactory(factory func() (frameSource, error)) {
manager.frameSourceFn = factory
}

func (manager *StreamSinkManagerCtx) Bitrate() uint64 {
manager.listenersMu.Lock()
defer manager.listenersMu.Unlock()
Expand Down Expand Up @@ -325,9 +331,27 @@ func (manager *StreamSinkManagerCtx) CreatePipeline() error {
return err
}

if manager.frameSourceFn != nil {
manager.frameSource, err = manager.frameSourceFn()
if err != nil {
manager.pipeline.Destroy()
manager.pipeline = nil
return err
}
manager.pipeline.AttachAppsrc("appsrc")
}
manager.pipeline.AttachAppsink("appsink")
manager.pipeline.Play()

if manager.frameSource != nil {
if err := manager.frameSource.Start(manager.pipeline.Push); err != nil {
manager.pipeline.Destroy()
manager.pipeline = nil
manager.frameSource = nil
return err
}
}

manager.wg.Add(1)
pipeline := manager.pipeline

Expand Down Expand Up @@ -405,6 +429,10 @@ func (manager *StreamSinkManagerCtx) DestroyPipeline() {
return
}

if manager.frameSource != nil {
manager.frameSource.Stop()
manager.frameSource = nil
}
manager.pipeline.Destroy()
manager.logger.Info().Msgf("destroying pipeline")
manager.pipeline = nil
Expand Down
136 changes: 136 additions & 0 deletions server/internal/capture/wayland.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,136 @@
package capture

import (
"context"
"fmt"
"io"
"os"
"os/exec"
"strconv"
"sync"

"github.com/rs/zerolog/log"

"github.com/m1k1o/neko/server/pkg/types"
)

type frameSource interface {
Start(func([]byte)) error
Stop()
}

type waylandFrameSource struct {
recorder string
width int
height int
fps int

mu sync.Mutex
cancel context.CancelFunc
done chan struct{}
}

func newWaylandFrameSource(recorder string, screen types.ScreenSize) *waylandFrameSource {
fps := int(screen.Rate)
if fps <= 0 {
fps = 25
}

return &waylandFrameSource{
recorder: recorder,
width: screen.Width,
height: screen.Height,
fps: fps,
}
}

func (source *waylandFrameSource) frameSize() int {
return source.width * source.height * 4
}

func (source *waylandFrameSource) args() []string {
return []string{
"--no-damage",
"--no-dmabuf",
"--framerate", strconv.Itoa(source.fps),
"--muxer", "rawvideo",
"--codec", "rawvideo",
"--pixel-format", "bgr0",
"--file", "/dev/stdout",
"--overwrite",
}
}

func (source *waylandFrameSource) command() *exec.Cmd {
return exec.Command(source.recorder, source.args()...)
}

func (source *waylandFrameSource) Start(push func([]byte)) error {
if push == nil {
return fmt.Errorf("frame push callback is required")
}
if source.recorder == "" {
return fmt.Errorf("Wayland recorder executable is required")
}
if source.width <= 0 || source.height <= 0 {
return fmt.Errorf("invalid Wayland output size: %dx%d", source.width, source.height)
}

ctx, cancel := context.WithCancel(context.Background())
cmd := exec.CommandContext(ctx, source.recorder, source.args()...)
cmd.Stderr = os.Stderr

stdout, err := cmd.StdoutPipe()
if err != nil {
cancel()
return fmt.Errorf("create Wayland recorder pipe: %w", err)
}
if err := cmd.Start(); err != nil {
cancel()
return fmt.Errorf("start Wayland recorder: %w", err)
}

done := make(chan struct{})
source.mu.Lock()
source.cancel = cancel
source.done = done
source.mu.Unlock()

go func() {
defer close(done)
defer stdout.Close()

frame := make([]byte, source.frameSize())
for {
if _, err := io.ReadFull(stdout, frame); err != nil {
if err != io.EOF && err != io.ErrUnexpectedEOF {
log.Warn().Err(err).Msg("Wayland recorder stopped while reading a frame")
}
break
}

push(frame)
}

if err := cmd.Wait(); err != nil && ctx.Err() == nil {
log.Warn().Err(err).Msg("Wayland recorder exited")
}
}()

return nil
}

func (source *waylandFrameSource) Stop() {
source.mu.Lock()
cancel := source.cancel
done := source.done
source.cancel = nil
source.done = nil
source.mu.Unlock()

if cancel == nil {
return
}
cancel()
<-done
}
46 changes: 46 additions & 0 deletions server/internal/capture/wayland_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
package capture

import (
"reflect"
"testing"

"github.com/m1k1o/neko/server/pkg/types"
)

func TestWaylandFrameSourceCommand(t *testing.T) {
source := newWaylandFrameSource("wf-recorder", types.ScreenSize{
Width: 1920,
Height: 1080,
Rate: 25,
})

got := source.command().Args
want := []string{
"wf-recorder",
"--no-damage",
"--no-dmabuf",
"--framerate", "25",
"--muxer", "rawvideo",
"--codec", "rawvideo",
"--pixel-format", "bgr0",
"--file", "/dev/stdout",
"--overwrite",
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("command args = %#v, want %#v", got, want)
}
}

func TestWaylandFrameSourceDefaultsFrameRate(t *testing.T) {
source := newWaylandFrameSource("wf-recorder", types.ScreenSize{
Width: 10,
Height: 20,
})

if source.fps != 25 {
t.Fatalf("fps = %d, want 25", source.fps)
}
if source.frameSize() != 800 {
t.Fatalf("frame size = %d, want 800", source.frameSize())
}
}
16 changes: 16 additions & 0 deletions server/internal/config/capture.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,9 @@ const (
type Capture struct {
Display string

Wayland bool
WaylandRecorder string

VideoCodec codec.RTPCodec
VideoIDs []string
VideoPipelines map[string]types.VideoConfig
Expand Down Expand Up @@ -80,6 +83,16 @@ func (Capture) Init(cmd *cobra.Command) error {
return err
}

cmd.PersistentFlags().Bool("capture.video.wayland", false, "capture a Wayland compositor output")
if err := viper.BindPFlag("capture.video.wayland", cmd.PersistentFlags().Lookup("capture.video.wayland")); err != nil {
return err
}

cmd.PersistentFlags().String("capture.video.wayland_recorder", "wf-recorder", "Wayland screencopy recorder executable")
if err := viper.BindPFlag("capture.video.wayland_recorder", cmd.PersistentFlags().Lookup("capture.video.wayland_recorder")); err != nil {
return err
}

cmd.PersistentFlags().String("capture.video.codec", "vp8", "video codec to be used")
if err := viper.BindPFlag("capture.video.codec", cmd.PersistentFlags().Lookup("capture.video.codec")); err != nil {
return err
Expand Down Expand Up @@ -326,6 +339,9 @@ func (s *Capture) Set() {
s.Display = os.Getenv("DISPLAY")
}

s.Wayland = viper.GetBool("capture.video.wayland")
s.WaylandRecorder = viper.GetString("capture.video.wayland_recorder")

// video
videoCodec := viper.GetString("capture.video.codec")
s.VideoCodec, ok = codec.ParseStr(videoCodec)
Expand Down
23 changes: 22 additions & 1 deletion server/internal/config/desktop.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@ import (
)

type Desktop struct {
Display string
Display string
Wayland bool
WaylandOutput string
WaylandResizeCommand string

ScreenSize types.ScreenSize

Expand All @@ -31,6 +34,21 @@ func (Desktop) Init(cmd *cobra.Command) error {
return err
}

cmd.PersistentFlags().Bool("desktop.wayland", false, "use Wayland desktop input and screen management")
if err := viper.BindPFlag("desktop.wayland", cmd.PersistentFlags().Lookup("desktop.wayland")); err != nil {
return err
}

cmd.PersistentFlags().String("desktop.wayland.output", "HEADLESS-1", "Wayland output name used for resizing")
if err := viper.BindPFlag("desktop.wayland.output", cmd.PersistentFlags().Lookup("desktop.wayland.output")); err != nil {
return err
}

cmd.PersistentFlags().String("desktop.wayland.resize_command", "wlr-randr", "Wayland output resize executable")
if err := viper.BindPFlag("desktop.wayland.resize_command", cmd.PersistentFlags().Lookup("desktop.wayland.resize_command")); err != nil {
return err
}

cmd.PersistentFlags().String("desktop.screen", "1280x720@30", "default screen size and framerate")
if err := viper.BindPFlag("desktop.screen", cmd.PersistentFlags().Lookup("desktop.screen")); err != nil {
return err
Expand Down Expand Up @@ -75,6 +93,9 @@ func (Desktop) InitV2(cmd *cobra.Command) error {

func (s *Desktop) Set() {
s.Display = viper.GetString("desktop.display")
s.Wayland = viper.GetBool("desktop.wayland")
s.WaylandOutput = viper.GetString("desktop.wayland.output")
s.WaylandResizeCommand = viper.GetString("desktop.wayland.resize_command")

// Display is provided by env variable unless explicitly set
if s.Display == "" {
Expand Down
Loading
Loading