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 { logger zerolog.Logger desktop types.DesktopManager // sinks broadcast *BroacastManagerCtx audio *StreamSinkManagerCtx video *StreamSinkManagerCtx // source-sinks screenshare *StreamSrcSinkManagerCtx } func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCtx { logger := log.With().Str("module", "capture").Logger() return &CaptureManagerCtx{ logger: logger, desktop: desktop, // sinks broadcast: broadcastNew(func(url string) (string, error) { return NewBroadcastPipeline(config.AudioDevice, config.Display, config.BroadcastPipeline, url) }, config.BroadcastUrl), audio: streamSinkNew(config.AudioCodec, func() (string, error) { return NewAudioPipeline(config.AudioCodec, config.AudioDevice, config.AudioPipeline, config.AudioBitrate) }, "audio"), video: streamSinkNew(config.VideoCodec, func() (string, error) { // use screen fps as default fps := desktop.GetScreenSize().Rate // if max fps is set, cap it to that value if config.VideoMaxFPS > 0 && config.VideoMaxFPS < fps { fps = config.VideoMaxFPS } 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"), } } func (manager *CaptureManagerCtx) Start() { if manager.broadcast.Started() { if err := manager.broadcast.createPipeline(); err != nil { manager.logger.Panic().Err(err).Msg("unable to create broadcast pipeline") } } go func() { for { before, ok := <-manager.desktop.GetScreenSizeChangeChannel() if !ok { manager.logger.Info().Msg("screen size change channel was closed") return } if before { // before screen size change, we need to destroy all pipelines if manager.video.Started() { manager.video.destroyPipeline() } if manager.broadcast.Started() { manager.broadcast.destroyPipeline() } } else { // after screen size change, we need to recreate all pipelines if manager.video.Started() { err := manager.video.createPipeline() if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) { manager.logger.Panic().Err(err).Msg("unable to recreate video pipeline") } } if manager.broadcast.Started() { err := manager.broadcast.createPipeline() if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) { manager.logger.Panic().Err(err).Msg("unable to recreate broadcast pipeline") } } } } }() } func (manager *CaptureManagerCtx) Shutdown() error { manager.logger.Info().Msgf("shutdown") manager.screenshare.shutdown() manager.broadcast.shutdown() manager.audio.shutdown() manager.video.shutdown() return nil } func (manager *CaptureManagerCtx) Broadcast() types.BroadcastManager { return manager.broadcast } func (manager *CaptureManagerCtx) Audio() types.StreamSinkManager { return manager.audio } func (manager *CaptureManagerCtx) Video() types.StreamSinkManager { return manager.video } func (manager *CaptureManagerCtx) Screenshare() types.StreamSrcSinkManager { return manager.screenshare }