pipelineFn returns string.

server/internal/capture/broadcast.go

@@ -16,13 +16,13 @@ type BroacastManagerCtx struct {
 
 	pipeline   *gst.Pipeline
 	pipelineMu sync.Mutex
-	pipelineFn func(url string) (*gst.Pipeline, error)
+	pipelineFn func(url string) (string, error)
 
 	url     string
 	started bool
 }
 
-func broadcastNew(pipelineFn func(url string) (*gst.Pipeline, error), url string, started bool) *BroacastManagerCtx {
+func broadcastNew(pipelineFn func(url string) (string, error), url string, started bool) *BroacastManagerCtx {
 	logger := log.With().
 		Str("module", "capture").
 		Str("submodule", "broadcast").
@@ -87,20 +87,20 @@ func (manager *BroacastManagerCtx) createPipeline() error {
 	}
 
 	var err error
-
-	manager.logger.Info().
-		Str("url", manager.url).
-		Msgf("creating pipeline")
-
-	manager.pipeline, err = manager.pipelineFn(manager.url)
+	pipelineStr, err := manager.pipelineFn(manager.url)
 	if err != nil {
 		return err
 	}
 
 	manager.logger.Info().
 		Str("url", manager.url).
-		Str("src", manager.pipeline.Src).
-		Msgf("created pipeline")
+		Str("src", pipelineStr).
+		Msgf("starting pipeline")
+
+	manager.pipeline, err = gst.CreatePipeline(pipelineStr)
+	if err != nil {
+		return err
+	}
 
 	manager.pipeline.Play()
 

server/internal/capture/manager.go

@@ -6,7 +6,6 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
-	"m1k1o/neko/internal/capture/gst"
 	"m1k1o/neko/internal/config"
 	"m1k1o/neko/internal/types"
 )
@@ -29,13 +28,13 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 		desktop: desktop,
 
 		// sinks
-		broadcast: broadcastNew(func(url string) (*gst.Pipeline, error) {
+		broadcast: broadcastNew(func(url string) (string, error) {
 			return NewBroadcastPipeline(config.AudioDevice, config.Display, config.BroadcastPipeline, url)
 		}, config.BroadcastUrl, config.BroadcastStarted),
-		audio: streamSinkNew(config.AudioCodec, func() (*gst.Pipeline, error) {
+		audio: streamSinkNew(config.AudioCodec, func() (string, error) {
 			return NewAudioPipeline(config.AudioCodec, config.AudioDevice, config.AudioPipeline, config.AudioBitrate)
 		}, "audio"),
-		video: streamSinkNew(config.VideoCodec, func() (*gst.Pipeline, error) {
+		video: streamSinkNew(config.VideoCodec, func() (string, error) {
 			return NewVideoPipeline(config.VideoCodec, config.Display, config.VideoPipeline, config.VideoMaxFPS, config.VideoBitrate, config.VideoHWEnc)
 		}, "audio"),
 	}

server/internal/capture/pipelines.go

@@ -33,7 +33,7 @@ const (
 	audioSrc = "pulsesrc device=%s ! audio/x-raw,channels=2 ! audioconvert ! "
 )
 
-func NewBroadcastPipeline(device string, display string, pipelineSrc string, url string) (*gst.Pipeline, error) {
+func NewBroadcastPipeline(device string, display string, pipelineSrc string, url string) (string, error) {
 	video := fmt.Sprintf(videoSrc, display, 25)
 	audio := fmt.Sprintf(audioSrc, device)
 
@@ -49,23 +49,23 @@ func NewBroadcastPipeline(device string, display string, pipelineSrc string, url
 		pipelineStr = fmt.Sprintf("flvmux name=mux ! rtmpsink location='%s live=1' %s audio/x-raw,channels=2 ! audioconvert ! voaacenc ! mux. %s x264enc bframes=0 key-int-max=60 byte-stream=true tune=zerolatency speed-preset=veryfast ! mux.", url, audio, video)
 	}
 
-	return gst.CreatePipeline(pipelineStr)
+	return pipelineStr, nil
 }
 
-func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc string, fps int16, bitrate uint, hwenc string) (*gst.Pipeline, error) {
+func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc string, fps int16, bitrate uint, hwenc string) (string, error) {
 	pipelineStr := " ! appsink name=appsink"
 
 	// if using custom pipeline
 	if pipelineSrc != "" {
 		pipelineStr = fmt.Sprintf(pipelineSrc+pipelineStr, display)
-		return gst.CreatePipeline(pipelineStr)
+		return pipelineStr, nil
 	}
 
 	switch rtpCodec.Name {
 	case codec.VP8().Name:
 		if hwenc == "VAAPI" {
 			if err := gst.CheckPlugins([]string{"ximagesrc", "vaapi"}); err != nil {
-				return nil, err
+				return "", err
 			}
 			// vp8 encode is missing from gstreamer.freedesktop.org/documentation
 			// note that it was removed from some recent intel CPUs: https://trac.ffmpeg.org/wiki/Hardware/QuickSync
@@ -76,7 +76,7 @@ func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc strin
 			// gstreamer1.0-plugins-good
 			// vp8enc error-resilient=partitions keyframe-max-dist=10 auto-alt-ref=true cpu-used=5 deadline=1
 			if err := gst.CheckPlugins([]string{"ximagesrc", "vpx"}); err != nil {
-				return nil, err
+				return "", err
 			}
 
 			pipelineStr = strings.Join([]string{
@@ -102,18 +102,18 @@ func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc strin
 		// gstreamer1.0-plugins-good
 		// vp9enc
 		if err := gst.CheckPlugins([]string{"ximagesrc", "vpx"}); err != nil {
-			return nil, err
+			return "", err
 		}
 
 		pipelineStr = fmt.Sprintf(videoSrc+"vp9enc target-bitrate=%d cpu-used=-5 threads=4 deadline=1 keyframe-max-dist=30 auto-alt-ref=true"+pipelineStr, display, fps, bitrate*1000)
 	case codec.H264().Name:
 		if err := gst.CheckPlugins([]string{"ximagesrc"}); err != nil {
-			return nil, err
+			return "", err
 		}
 
 		if hwenc == "VAAPI" {
 			if err := gst.CheckPlugins([]string{"vaapi"}); err != nil {
-				return nil, err
+				return "", err
 			}
 
 			pipelineStr = fmt.Sprintf(videoSrc+"video/x-raw,format=NV12 ! vaapih264enc rate-control=vbr bitrate=%d keyframe-period=180 quality-level=7 ! video/x-h264,stream-format=byte-stream"+pipelineStr, display, fps, bitrate)
@@ -131,7 +131,7 @@ func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc strin
 			// gstreamer1.0-plugins-ugly
 			// video/x-raw,format=I420 ! x264enc bframes=0 key-int-max=60 byte-stream=true tune=zerolatency speed-preset=veryfast ! video/x-h264,stream-format=byte-stream
 			if err := gst.CheckPlugins([]string{"x264"}); err != nil {
-				return nil, err
+				return "", err
 			}
 
 			vbvbuf := uint(1000)
@@ -142,19 +142,19 @@ func NewVideoPipeline(rtpCodec codec.RTPCodec, display string, pipelineSrc strin
 			pipelineStr = fmt.Sprintf(videoSrc+"video/x-raw,format=NV12 ! x264enc threads=4 bitrate=%d key-int-max=60 vbv-buf-capacity=%d byte-stream=true tune=zerolatency speed-preset=veryfast ! video/x-h264,stream-format=byte-stream"+pipelineStr, display, fps, bitrate, vbvbuf)
 		}
 	default:
-		return nil, fmt.Errorf("unknown codec %s", rtpCodec.Name)
+		return "", fmt.Errorf("unknown codec %s", rtpCodec.Name)
 	}
 
-	return gst.CreatePipeline(pipelineStr)
+	return pipelineStr, nil
 }
 
-func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string, bitrate uint) (*gst.Pipeline, error) {
+func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string, bitrate uint) (string, error) {
 	pipelineStr := " ! appsink name=appsink"
 
 	// if using custom pipeline
 	if pipelineSrc != "" {
 		pipelineStr = fmt.Sprintf(pipelineSrc+pipelineStr, device)
-		return gst.CreatePipeline(pipelineStr)
+		return pipelineStr, nil
 	}
 
 	switch rtpCodec.Name {
@@ -163,7 +163,7 @@ func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string
 		// gstreamer1.0-plugins-base
 		// opusenc
 		if err := gst.CheckPlugins([]string{"pulseaudio", "opus"}); err != nil {
-			return nil, err
+			return "", err
 		}
 
 		pipelineStr = fmt.Sprintf(audioSrc+"opusenc inband-fec=true bitrate=%d"+pipelineStr, device, bitrate*1000)
@@ -172,7 +172,7 @@ func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string
 		// gstreamer1.0-libav
 		// avenc_g722
 		if err := gst.CheckPlugins([]string{"pulseaudio", "libav"}); err != nil {
-			return nil, err
+			return "", err
 		}
 
 		pipelineStr = fmt.Sprintf(audioSrc+"avenc_g722 bitrate=%d"+pipelineStr, device, bitrate*1000)
@@ -181,7 +181,7 @@ func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string
 		// gstreamer1.0-plugins-good
 		// audio/x-raw, rate=8000 ! mulawenc
 		if err := gst.CheckPlugins([]string{"pulseaudio", "mulaw"}); err != nil {
-			return nil, err
+			return "", err
 		}
 
 		pipelineStr = fmt.Sprintf(audioSrc+"audio/x-raw, rate=8000 ! mulawenc"+pipelineStr, device)
@@ -190,13 +190,13 @@ func NewAudioPipeline(rtpCodec codec.RTPCodec, device string, pipelineSrc string
 		// gstreamer1.0-plugins-good
 		// audio/x-raw, rate=8000 ! alawenc
 		if err := gst.CheckPlugins([]string{"pulseaudio", "alaw"}); err != nil {
-			return nil, err
+			return "", err
 		}
 
 		pipelineStr = fmt.Sprintf(audioSrc+"audio/x-raw, rate=8000 ! alawenc"+pipelineStr, device)
 	default:
-		return nil, fmt.Errorf("unknown codec %s", rtpCodec.Name)
+		return "", fmt.Errorf("unknown codec %s", rtpCodec.Name)
 	}
 
-	return gst.CreatePipeline(pipelineStr)
+	return pipelineStr, nil
 }

server/internal/capture/streamsink.go

@@ -12,8 +12,6 @@ import (
 	"m1k1o/neko/internal/types/codec"
 )
 
-var moveSinkListenerMu = sync.Mutex{}
-
 type StreamSinkManagerCtx struct {
 	logger zerolog.Logger
 	mu     sync.Mutex
@@ -22,7 +20,7 @@ type StreamSinkManagerCtx struct {
 	codec      codec.RTPCodec
 	pipeline   *gst.Pipeline
 	pipelineMu sync.Mutex
-	pipelineFn func() (*gst.Pipeline, error)
+	pipelineFn func() (string, error)
 
 	listeners   int
 	listenersMu sync.Mutex
@@ -30,7 +28,7 @@ type StreamSinkManagerCtx struct {
 	sampleFn func(sample types.Sample)
 }
 
-func streamSinkNew(codec codec.RTPCodec, pipelineFn func() (*gst.Pipeline, error), video_id string) *StreamSinkManagerCtx {
+func streamSinkNew(codec codec.RTPCodec, pipelineFn func() (string, error), video_id string) *StreamSinkManagerCtx {
 	logger := log.With().
 		Str("module", "capture").
 		Str("submodule", "stream-sink").
@@ -139,21 +137,20 @@ func (manager *StreamSinkManagerCtx) createPipeline() error {
 		return types.ErrCapturePipelineAlreadyExists
 	}
 
-	var err error
-
-	manager.logger.Info().
-		Str("codec", manager.codec.Name).
-		Msgf("creating pipeline")
-
-	manager.pipeline, err = manager.pipelineFn()
+	pipelineStr, err := manager.pipelineFn()
 	if err != nil {
 		return err
 	}
 
 	manager.logger.Info().
 		Str("codec", manager.codec.Name).
-		Str("src", manager.pipeline.Src).
-		Msgf("created pipeline")
+		Str("src", pipelineStr).
+		Msgf("creating pipeline")
+
+	manager.pipeline, err = gst.CreatePipeline(pipelineStr)
+	if err != nil {
+		return err
+	}
 
 	manager.pipeline.AttachAppsink("appsink")
 	manager.pipeline.Play()