screenshare POC.

server/internal/capture/manager.go

@@ -2,12 +2,14 @@ package capture
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
 	"m1k1o/neko/internal/config"
 	"m1k1o/neko/internal/types"
+	"m1k1o/neko/internal/types/codec"
 )
 
 type CaptureManagerCtx struct {
@@ -18,6 +20,9 @@ type CaptureManagerCtx struct {
 	broadcast *BroacastManagerCtx
 	audio     *StreamSinkManagerCtx
 	video     *StreamSinkManagerCtx
+
+	// source-sinks
+	screenshare *StreamSrcSinkManagerCtx
 }
 
 func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCtx {
@@ -43,6 +48,15 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 			}
 			return NewVideoPipeline(config.VideoCodec, config.Display, config.VideoPipeline, fps, config.VideoBitrate, config.VideoHWEnc)
 		}, "video"),
+
+		// source-sinks
+		screenshare: streamSrcSinkNew(config.ScreenshareEnabled, map[string]string{
+			codec.VP8().Name: "appsrc format=time is-live=true do-timestamp=true name=appsrc " +
+				fmt.Sprintf("! application/x-rtp, payload=%d, encoding-name=VP8-DRAFT-IETF-01 ", codec.VP8().PayloadType) +
+				"! rtpvp8depay " +
+				"! appsink name=appsink",
+			// TODO: Add support for more codecs.
+		}, "webcam"),
 	}
 }
 
@@ -95,6 +109,7 @@ func (manager *CaptureManagerCtx) Start() {
 func (manager *CaptureManagerCtx) Shutdown() error {
 	manager.logger.Info().Msgf("shutdown")
 
+	manager.screenshare.shutdown()
 	manager.broadcast.shutdown()
 
 	manager.audio.shutdown()
@@ -114,3 +129,7 @@ func (manager *CaptureManagerCtx) Audio() types.StreamSinkManager {
 func (manager *CaptureManagerCtx) Video() types.StreamSinkManager {
 	return manager.video
 }
+
+func (manager *CaptureManagerCtx) Screenshare() types.StreamSrcSinkManager {
+	return manager.screenshare
+}

server/internal/capture/streamsrcsink.go

@@ -0,0 +1,137 @@
+package capture
+
+import (
+	"errors"
+	"sync"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+
+	"m1k1o/neko/internal/capture/gst"
+	"m1k1o/neko/internal/types"
+	"m1k1o/neko/internal/types/codec"
+)
+
+type StreamSrcSinkManagerCtx struct {
+	logger        zerolog.Logger
+	sampleChannel chan types.Sample
+
+	enabled       bool
+	codecPipeline map[string]string // codec -> pipeline
+
+	codec       codec.RTPCodec
+	pipeline    *gst.Pipeline
+	pipelineMu  sync.Mutex
+	pipelineStr string
+}
+
+func streamSrcSinkNew(enabled bool, codecPipeline map[string]string, video_id string) *StreamSrcSinkManagerCtx {
+	logger := log.With().
+		Str("module", "capture").
+		Str("submodule", "stream-src-sink").
+		Str("video_id", video_id).Logger()
+
+	return &StreamSrcSinkManagerCtx{
+		logger:        logger,
+		enabled:       enabled,
+		codecPipeline: codecPipeline,
+		sampleChannel: make(chan types.Sample),
+	}
+}
+
+func (manager *StreamSrcSinkManagerCtx) shutdown() {
+	manager.logger.Info().Msgf("shutdown")
+
+	manager.Stop()
+}
+
+func (manager *StreamSrcSinkManagerCtx) Codec() codec.RTPCodec {
+	manager.pipelineMu.Lock()
+	defer manager.pipelineMu.Unlock()
+
+	return manager.codec
+}
+
+func (manager *StreamSrcSinkManagerCtx) Start(codec codec.RTPCodec) error {
+	manager.pipelineMu.Lock()
+	defer manager.pipelineMu.Unlock()
+
+	if manager.pipeline != nil {
+		return types.ErrCapturePipelineAlreadyExists
+	}
+
+	if !manager.enabled {
+		return errors.New("stream-src-sink not enabled")
+	}
+
+	found := false
+	for codecName, pipeline := range manager.codecPipeline {
+		if codecName == codec.Name {
+			manager.pipelineStr = pipeline
+			manager.codec = codec
+			found = true
+			break
+		}
+	}
+
+	if !found {
+		return errors.New("no pipeline found for a codec")
+	}
+
+	var err error
+
+	manager.logger.Info().
+		Str("codec", manager.codec.Name).
+		Str("src", manager.pipelineStr).
+		Msgf("creating pipeline")
+
+	manager.pipeline, err = gst.CreatePipeline(manager.pipelineStr)
+	if err != nil {
+		return err
+	}
+
+	manager.pipeline.AttachAppsrc("appsrc")
+	manager.pipeline.AttachAppsink("appsink", manager.sampleChannel)
+	manager.pipeline.Play()
+
+	return nil
+}
+
+func (manager *StreamSrcSinkManagerCtx) Stop() {
+	manager.pipelineMu.Lock()
+	defer manager.pipelineMu.Unlock()
+
+	if manager.pipeline == nil {
+		return
+	}
+
+	manager.pipeline.Destroy()
+	manager.pipeline = nil
+
+	manager.logger.Info().
+		Str("codec", manager.codec.Name).
+		Str("src", manager.pipelineStr).
+		Msgf("destroying pipeline")
+}
+
+func (manager *StreamSrcSinkManagerCtx) Push(bytes []byte) {
+	manager.pipelineMu.Lock()
+	defer manager.pipelineMu.Unlock()
+
+	if manager.pipeline == nil {
+		return
+	}
+
+	manager.pipeline.Push(bytes)
+}
+
+func (manager *StreamSrcSinkManagerCtx) Started() bool {
+	manager.pipelineMu.Lock()
+	defer manager.pipelineMu.Unlock()
+
+	return manager.pipeline != nil
+}
+
+func (manager *StreamSrcSinkManagerCtx) GetSampleChannel() chan types.Sample {
+	return manager.sampleChannel
+}