add screencast to capture.

internal/api/room/handler.go

@@ -63,6 +63,7 @@ func (h *RoomHandler) Route(r chi.Router) {
 	r.Route("/screen", func(r chi.Router) {
 		r.With(auth.CanWatchOnly).Get("/", h.screenConfiguration)
 		r.With(auth.CanWatchOnly).Get("/image", h.screenImageGet)
+		r.With(auth.CanWatchOnly).Get("/cast", h.screenCastGet)
 
 		r.With(auth.AdminsOnly).Post("/", h.screenConfigurationChange)
 		r.With(auth.AdminsOnly).Get("/configurations", h.screenConfigurationsList)

internal/api/room/screen.go

@@ -95,3 +95,16 @@ func (h *RoomHandler) screenImageGet(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "image/jpeg")
 	w.Write(out.Bytes())
 }
+
+func (h *RoomHandler) screenCastGet(w http.ResponseWriter, r *http.Request) {
+	screencast := h.capture.Screencast()
+	if !screencast.Enabled() {
+		utils.HttpBadRequest(w, "Screencast pipeline is not enabled.")
+		return
+	}
+
+	bytes := screencast.Image()
+	w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
+	w.Header().Set("Content-Type", "image/jpeg")
+	w.Write(bytes)
+}

internal/capture/gst/gst.go

@@ -58,6 +58,18 @@ func CreateRTMPPipeline(pipelineDevice string, pipelineDisplay string, pipelineS
 	return CreatePipeline(pipelineStr, 0)
 }
 
+// CreateJPEGPipeline creates a GStreamer Pipeline
+func CreateJPEGPipeline(pipelineDisplay string, pipelineSrc string) (*Pipeline, error) {
+	var pipelineStr string
+	if pipelineSrc != "" {
+		pipelineStr = fmt.Sprintf(pipelineSrc, pipelineDisplay)
+	} else {
+		pipelineStr = fmt.Sprintf("ximagesrc display-name=%s show-pointer=true use-damage=false ! videoconvert ! videoscale ! videorate ! video/x-raw,framerate=10/1 ! jpegenc quality=60" + appSink, pipelineDisplay)
+	}
+
+	return CreatePipeline(pipelineStr, 0)
+}
+
 // CreateAppPipeline creates a GStreamer Pipeline
 func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc string) (*Pipeline, error) {
 	var clockRate float32
@@ -222,15 +234,18 @@ func CheckPlugins(plugins []string) error {
 
 //export goHandlePipelineBuffer
 func goHandlePipelineBuffer(buffer unsafe.Pointer, bufferLen C.int, duration C.int, pipelineID C.int) {
+	defer C.free(buffer)
+
 	pipelinesLock.Lock()
 	pipeline, ok := pipelines[int(pipelineID)]
 	pipelinesLock.Unlock()
 
 	if ok {
-		samples := uint32(pipeline.ClockRate * (float32(duration) / 1000000000))
+		pipeline.Sample <- types.Sample{
-		pipeline.Sample <- types.Sample{Data: C.GoBytes(buffer, bufferLen), Samples: samples}
+			Data: C.GoBytes(buffer, bufferLen),
+			Samples: uint32(pipeline.ClockRate * (float32(duration) / 1e9)),
+		}
 	} else {
 		fmt.Printf("discarding buffer, no pipeline with id %d", int(pipelineID))
 	}
-	C.free(buffer)
 }

internal/capture/manager.go

@@ -26,6 +26,7 @@ type CaptureManagerCtx struct {
 	streaming       bool
 	desktop         types.DesktopManager
 	broadcast       *BroacastManagerCtx
+	screencast      *ScreencastManagerCtx
 }
 
 func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCtx {
@@ -39,6 +40,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 		streaming:       false,
 		desktop:         desktop,
 		broadcast:       broadcastNew(config),
+		screencast:      screencastNew(config),
 	}
 }
 
@@ -49,6 +51,12 @@ func (manager *CaptureManagerCtx) Start() {
 		}
 	}
 
+	if manager.screencast.Enabled() {
+		if err := manager.screencast.createPipeline(); err != nil {
+			manager.logger.Panic().Err(err).Msg("unable to create screencast pipeline")
+		}
+	}
+
 	manager.desktop.OnBeforeScreenSizeChange(func() {
 		if manager.Streaming() {
 			manager.destroyVideoPipeline()
@@ -57,6 +65,10 @@ func (manager *CaptureManagerCtx) Start() {
 		if manager.broadcast.Enabled() {
 			manager.broadcast.destroyPipeline()
 		}
+
+		if manager.screencast.Enabled() {
+			manager.screencast.destroyPipeline()
+		}
 	})
 
 	manager.desktop.OnAfterScreenSizeChange(func() {
@@ -66,7 +78,13 @@ func (manager *CaptureManagerCtx) Start() {
 
 		if manager.broadcast.Enabled() {
 			if err := manager.broadcast.createPipeline(); err != nil {
-				manager.logger.Panic().Err(err).Msg("unable to create broadcast pipeline")
+				manager.logger.Panic().Err(err).Msg("unable to recreate broadcast pipeline")
+			}
+		}
+
+		if manager.screencast.Enabled() {
+			if err := manager.screencast.createPipeline(); err != nil {
+				manager.logger.Panic().Err(err).Msg("unable to recreate screencast pipeline")
 			}
 		}
 	})
@@ -97,11 +115,11 @@ func (manager *CaptureManagerCtx) Shutdown() error {
 		manager.StopStream()
 	}
 
-	if manager.broadcast.Enabled() {
+	manager.broadcast.destroyPipeline()
-		manager.broadcast.destroyPipeline()
+	manager.screencast.destroyPipeline()
-	}
 
 	manager.emit_stop <- true
+	manager.screencast.shutdown <- true
 	return nil
 }
 
@@ -109,6 +127,10 @@ func (manager *CaptureManagerCtx) Broadcast() types.BroadcastManager {
 	return manager.broadcast
 }
 
+func (manager *CaptureManagerCtx) Screencast() types.ScreencastManager {
+	return manager.screencast
+}
+
 func (manager *CaptureManagerCtx) VideoCodec() string {
 	return manager.config.VideoCodec
 }

internal/capture/screencast.go

@@ -0,0 +1,106 @@
+package capture
+
+import (
+	"bytes"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+
+	"demodesk/neko/internal/config"
+	"demodesk/neko/internal/types"
+	"demodesk/neko/internal/capture/gst"
+)
+
+type ScreencastManagerCtx struct {
+	logger    zerolog.Logger
+	config    *config.Capture
+	pipeline  *gst.Pipeline
+	enabled   bool
+	shutdown  chan bool
+	refresh   chan bool
+	sample    chan types.Sample
+	image     *bytes.Buffer
+}
+
+func screencastNew(config *config.Capture) *ScreencastManagerCtx {
+	manager := &ScreencastManagerCtx{
+		logger:    log.With().Str("module", "capture").Str("submodule", "screencast").Logger(),
+		config:    config,
+		enabled:   config.Screencast,
+		shutdown:  make(chan bool),
+		refresh:   make(chan bool),
+		image:     new(bytes.Buffer),
+	}
+
+	go func() {
+		manager.logger.Debug().Msg("subroutine started")
+
+		for {
+			select {
+			case <-manager.shutdown:
+				manager.logger.Debug().Msg("shutting down")
+				return
+			case <-manager.refresh:
+				manager.logger.Debug().Msg("subroutine updated")
+			case sample := <-manager.sample:
+				manager.image.Reset()
+				manager.image.Write(sample.Data)
+			}
+		}
+	}()
+
+	return manager
+}
+
+func (manager *ScreencastManagerCtx) Start() error {
+	manager.enabled = true
+	return manager.createPipeline()
+}
+
+func (manager *ScreencastManagerCtx) Stop() {
+	manager.enabled = false
+	manager.destroyPipeline()
+}
+
+func (manager *ScreencastManagerCtx) Enabled() bool {
+	return manager.enabled
+}
+
+func (manager *ScreencastManagerCtx) Image() []byte {
+	return manager.image.Bytes()
+}
+
+func (manager *ScreencastManagerCtx) createPipeline() error {
+	var err error
+
+	manager.logger.Info().
+		Str("video_display", manager.config.Display).
+		Str("screencast_pipeline", manager.config.ScreencastPipeline).
+		Msgf("creating pipeline")
+	
+	manager.pipeline, err = gst.CreateJPEGPipeline(
+		manager.config.Display,
+		manager.config.ScreencastPipeline,
+	)
+
+	if err != nil {
+		return err
+	}
+
+	manager.pipeline.Start()
+	manager.logger.Info().Msgf("starting pipeline")
+
+	manager.sample = manager.pipeline.Sample
+	manager.refresh <-true
+	return nil
+}
+
+func (manager *ScreencastManagerCtx) destroyPipeline() {
+	if manager.pipeline == nil {
+		return
+	}
+
+	manager.pipeline.Stop()
+	manager.logger.Info().Msgf("stopping pipeline")
+	manager.pipeline = nil
+}

internal/config/capture.go

@@ -13,7 +13,10 @@ type Capture struct {
 	AudioParams  string
 	VideoCodec   string
 	VideoParams  string
+
 	BroadcastPipeline string
+	Screencast bool
+	ScreencastPipeline string
 }
 
 func (Capture) Init(cmd *cobra.Command) error {
@@ -80,6 +83,17 @@ func (Capture) Init(cmd *cobra.Command) error {
 		return err
 	}
 
+	// screencast
+	cmd.PersistentFlags().Bool("screencast", false, "enable screencast")
+	if err := viper.BindPFlag("screencast", cmd.PersistentFlags().Lookup("screencast")); err != nil {
+		return err
+	}
+
+	cmd.PersistentFlags().String("screencast_pipeline", "", "custom screencast pipeline")
+	if err := viper.BindPFlag("screencast_pipeline", cmd.PersistentFlags().Lookup("screencast_pipeline")); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -110,5 +124,8 @@ func (s *Capture) Set() {
 	s.Display = viper.GetString("display")
 	s.VideoCodec = videoCodec
 	s.VideoParams = viper.GetString("video")
+
 	s.BroadcastPipeline = viper.GetString("broadcast_pipeline")
+	s.Screencast = viper.GetBool("screencast")
+	s.ScreencastPipeline = viper.GetString("screencast_pipeline")
 }

internal/types/capture.go

@@ -12,11 +12,19 @@ type BroadcastManager interface {
 	Url() string
 }
 
+type ScreencastManager interface {
+	Start() error
+	Stop()
+	Enabled() bool
+	Image() []byte
+}
+
 type CaptureManager interface {
 	Start()
 	Shutdown() error
 
 	Broadcast() BroadcastManager
+	Screencast() ScreencastManager
 
 	VideoCodec() string
 	AudioCodec() string