diff --git a/internal/api/room/handler.go b/internal/api/room/handler.go index f620e430..f00b2d56 100644 --- a/internal/api/room/handler.go +++ b/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) diff --git a/internal/api/room/screen.go b/internal/api/room/screen.go index ffd3b2ca..d06d4a4f 100644 --- a/internal/api/room/screen.go +++ b/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) +} diff --git a/internal/capture/gst/gst.go b/internal/capture/gst/gst.go index d14711ef..6989abdd 100644 --- a/internal/capture/gst/gst.go +++ b/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{Data: C.GoBytes(buffer, bufferLen), Samples: samples} + pipeline.Sample <- types.Sample{ + 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) } diff --git a/internal/capture/manager.go b/internal/capture/manager.go index 4b15fc69..ee0988e7 100644 --- a/internal/capture/manager.go +++ b/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 } diff --git a/internal/capture/screencast.go b/internal/capture/screencast.go new file mode 100644 index 00000000..c9e1d7d1 --- /dev/null +++ b/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 +} diff --git a/internal/config/capture.go b/internal/config/capture.go index 5d6ea659..ede2a616 100644 --- a/internal/config/capture.go +++ b/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") } diff --git a/internal/types/capture.go b/internal/types/capture.go index 3fd5bbe2..9bd90a7e 100644 --- a/internal/types/capture.go +++ b/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