package capture import ( "errors" "sync" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "github.com/rs/zerolog" "github.com/rs/zerolog/log" "gitlab.com/demodesk/neko/server/pkg/gst" "gitlab.com/demodesk/neko/server/pkg/types" "gitlab.com/demodesk/neko/server/pkg/types/codec" ) type StreamSrcManagerCtx struct { logger zerolog.Logger enabled bool codecPipeline map[string]string // codec -> pipeline codec codec.RTPCodec pipeline *gst.Pipeline pipelineMu sync.Mutex pipelineStr string // metrics pushedData map[string]prometheus.Summary pipelinesCounter map[string]prometheus.Counter pipelinesActive map[string]prometheus.Gauge } func streamSrcNew(enabled bool, codecPipeline map[string]string, video_id string) *StreamSrcManagerCtx { logger := log.With(). Str("module", "capture"). Str("submodule", "stream-src"). Str("video_id", video_id).Logger() pushedData := map[string]prometheus.Summary{} pipelinesCounter := map[string]prometheus.Counter{} pipelinesActive := map[string]prometheus.Gauge{} for codecName, pipeline := range codecPipeline { codec, ok := codec.ParseStr(codecName) if !ok { logger.Fatal(). Str("codec", codecName). Str("pipeline", pipeline). Msg("unknown codec name") } pushedData[codecName] = promauto.NewSummary(prometheus.SummaryOpts{ Name: "streamsrc_data_bytes", Namespace: "neko", Subsystem: "capture", Help: "Data pushed to a pipeline (in bytes).", ConstLabels: map[string]string{ "video_id": video_id, "codec_name": codec.Name, "codec_type": codec.Type.String(), }, }) pipelinesCounter[codecName] = promauto.NewCounter(prometheus.CounterOpts{ Name: "pipelines_total", Namespace: "neko", Subsystem: "capture", Help: "Total number of created pipelines.", ConstLabels: map[string]string{ "submodule": "streamsrc", "video_id": video_id, "codec_name": codec.Name, "codec_type": codec.Type.String(), }, }) pipelinesActive[codecName] = promauto.NewGauge(prometheus.GaugeOpts{ Name: "pipelines_active", Namespace: "neko", Subsystem: "capture", Help: "Total number of active pipelines.", ConstLabels: map[string]string{ "submodule": "streamsrc", "video_id": video_id, "codec_name": codec.Name, "codec_type": codec.Type.String(), }, }) } return &StreamSrcManagerCtx{ logger: logger, enabled: enabled, codecPipeline: codecPipeline, // metrics pushedData: pushedData, pipelinesCounter: pipelinesCounter, } } func (manager *StreamSrcManagerCtx) shutdown() { manager.logger.Info().Msgf("shutdown") manager.Stop() } func (manager *StreamSrcManagerCtx) Codec() codec.RTPCodec { manager.pipelineMu.Lock() defer manager.pipelineMu.Unlock() return manager.codec } func (manager *StreamSrcManagerCtx) 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 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.Play() manager.pipelinesCounter[manager.codec.Name].Inc() manager.pipelinesActive[manager.codec.Name].Set(1) return nil } func (manager *StreamSrcManagerCtx) Stop() { manager.pipelineMu.Lock() defer manager.pipelineMu.Unlock() if manager.pipeline == nil { return } manager.pipeline.Destroy() manager.logger.Info().Msgf("destroying pipeline") manager.pipeline = nil manager.pipelinesActive[manager.codec.Name].Set(0) } func (manager *StreamSrcManagerCtx) Push(bytes []byte) { manager.pipelineMu.Lock() defer manager.pipelineMu.Unlock() if manager.pipeline == nil { return } manager.pipeline.Push(bytes) manager.pushedData[manager.codec.Name].Observe(float64(len(bytes))) } func (manager *StreamSrcManagerCtx) Started() bool { manager.pipelineMu.Lock() defer manager.pipelineMu.Unlock() return manager.pipeline != nil }