neko/internal/capture/streamsrc.go
Miroslav Šedivý e37f5c13ca fix bug.
2022-07-04 18:03:35 +02:00

198 lines
4.6 KiB
Go

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,
pipelinesActive: pipelinesActive,
}
}
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.pipeline = nil
manager.logger.Info().
Str("codec", manager.codec.Name).
Str("src", manager.pipelineStr).
Msgf("destroying pipeline")
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
}