neko/server/internal/capture/broadcast.go

157 lines
3.2 KiB
Go
Raw Normal View History

2020-11-02 04:09:48 +13:00
package capture
import (
2021-02-15 02:40:17 +13:00
"sync"
2021-02-06 00:49:02 +13:00
2022-06-15 10:23:16 +12:00
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"github.com/demodesk/neko/pkg/gst"
"github.com/demodesk/neko/pkg/types"
2020-11-02 04:09:48 +13:00
)
type BroacastManagerCtx struct {
2021-10-01 07:06:43 +13:00
logger zerolog.Logger
mu sync.Mutex
2022-10-14 06:57:31 +13:00
pipeline gst.Pipeline
2022-09-18 04:00:25 +12:00
pipelineMu sync.Mutex
pipelineFn func(url string) (string, error)
2021-10-01 07:06:43 +13:00
url string
started bool
2022-06-15 10:23:16 +12:00
// metrics
pipelinesCounter prometheus.Counter
2022-06-19 11:03:16 +12:00
pipelinesActive prometheus.Gauge
}
2024-07-19 04:58:40 +12:00
func broadcastNew(pipelineFn func(url string) (string, error), defaultUrl string, autostart bool) *BroacastManagerCtx {
2021-10-01 07:06:43 +13:00
logger := log.With().
Str("module", "capture").
Str("submodule", "broadcast").
Logger()
return &BroacastManagerCtx{
2022-09-18 04:00:25 +12:00
logger: logger,
pipelineFn: pipelineFn,
url: defaultUrl,
2024-07-19 04:58:40 +12:00
started: defaultUrl != "" && autostart,
2022-06-15 10:23:16 +12:00
// metrics
pipelinesCounter: promauto.NewCounter(prometheus.CounterOpts{
Name: "pipelines_total",
Namespace: "neko",
2022-06-19 10:51:17 +12:00
Subsystem: "capture",
2022-06-15 10:23:16 +12:00
Help: "Total number of created pipelines.",
2022-06-19 10:51:17 +12:00
ConstLabels: map[string]string{
"submodule": "broadcast",
"video_id": "main",
"codec_name": "-",
"codec_type": "-",
},
2022-06-15 10:23:16 +12:00
}),
2022-06-19 11:03:16 +12:00
pipelinesActive: promauto.NewGauge(prometheus.GaugeOpts{
Name: "pipelines_active",
Namespace: "neko",
Subsystem: "capture",
Help: "Total number of active pipelines.",
ConstLabels: map[string]string{
"submodule": "broadcast",
"video_id": "main",
"codec_name": "-",
"codec_type": "-",
},
}),
}
}
2021-02-06 00:18:46 +13:00
func (manager *BroacastManagerCtx) shutdown() {
2021-09-02 10:00:29 +12:00
manager.logger.Info().Msgf("shutdown")
2021-02-06 00:18:46 +13:00
manager.destroyPipeline()
}
func (manager *BroacastManagerCtx) Start(url string) error {
2021-02-06 00:49:02 +13:00
manager.mu.Lock()
defer manager.mu.Unlock()
2021-02-06 00:41:02 +13:00
err := manager.createPipeline()
if err != nil {
return err
}
manager.url = url
2021-02-06 02:03:53 +13:00
manager.started = true
2021-02-06 00:41:02 +13:00
return nil
}
func (manager *BroacastManagerCtx) Stop() {
2021-02-06 00:49:02 +13:00
manager.mu.Lock()
defer manager.mu.Unlock()
2021-02-06 02:03:53 +13:00
manager.started = false
manager.destroyPipeline()
}
2021-02-06 02:03:53 +13:00
func (manager *BroacastManagerCtx) Started() bool {
2021-10-01 07:06:43 +13:00
manager.mu.Lock()
defer manager.mu.Unlock()
2021-02-06 02:03:53 +13:00
return manager.started
}
func (manager *BroacastManagerCtx) Url() string {
2021-10-01 07:06:43 +13:00
manager.mu.Lock()
defer manager.mu.Unlock()
return manager.url
}
func (manager *BroacastManagerCtx) createPipeline() error {
2021-10-01 07:06:43 +13:00
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
2021-02-06 00:49:02 +13:00
if manager.pipeline != nil {
2021-08-30 03:09:13 +12:00
return types.ErrCapturePipelineAlreadyExists
2021-02-06 00:49:02 +13:00
}
2022-09-18 04:00:25 +12:00
pipelineStr, err := manager.pipelineFn(manager.url)
if err != nil {
return err
}
2021-02-06 01:58:02 +13:00
2020-11-02 04:09:48 +13:00
manager.logger.Info().
2022-09-22 04:58:46 +12:00
Str("url", manager.url).
2021-08-29 06:15:54 +12:00
Str("src", pipelineStr).
2021-02-06 01:58:02 +13:00
Msgf("starting pipeline")
2020-11-02 04:09:48 +13:00
2021-02-06 01:58:02 +13:00
manager.pipeline, err = gst.CreatePipeline(pipelineStr)
2020-11-02 04:09:48 +13:00
if err != nil {
return err
2020-11-02 04:09:48 +13:00
}
manager.pipeline.Play()
2022-06-15 10:23:16 +12:00
manager.pipelinesCounter.Inc()
2022-06-19 11:03:16 +12:00
manager.pipelinesActive.Set(1)
2022-06-15 10:23:16 +12:00
return nil
2020-11-02 04:09:48 +13:00
}
func (manager *BroacastManagerCtx) destroyPipeline() {
2021-10-01 07:06:43 +13:00
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
if manager.pipeline == nil {
2020-11-02 04:09:48 +13:00
return
}
2021-12-06 06:16:26 +13:00
manager.pipeline.Destroy()
2021-02-06 01:58:02 +13:00
manager.logger.Info().Msgf("destroying pipeline")
manager.pipeline = nil
2022-06-19 11:03:16 +12:00
manager.pipelinesActive.Set(0)
2020-11-02 04:09:48 +13:00
}