iridium/neko
2
2
Fork 0
mirror of https://github.com/m1k1o/neko.git synced 2024-07-24 14:40:50 +12:00
Code Issues Packages Projects Releases Wiki Activity

gst: move sample channel to AttachAppsink.

Browse Source
This commit is contained in:
Miroslav Šedivý 2023-01-29 19:59:54 +01:00
parent fdf17cfe77
commit 009ceddbd3
3 changed files with 15 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/internal/capture/broadcast.go
View File

@ -13,7 +13,6 @@ import (
type BroacastManagerCtx struct { type BroacastManagerCtx struct {
logger zerolog.Logger logger zerolog.Logger
mu sync.Mutex mu sync.Mutex
sampleChannel chan types.Sample
pipeline *gst.Pipeline pipeline *gst.Pipeline
pipelineMu sync.Mutex pipelineMu sync.Mutex
@ -32,7 +31,6 @@ func broadcastNew(pipelineFn func(url string) (string, error), defaultUrl string
return &BroacastManagerCtx{ return &BroacastManagerCtx{
logger: logger, logger: logger,
pipelineFn: pipelineFn, pipelineFn: pipelineFn,
sampleChannel: make(chan types.Sample),
url: defaultUrl, url: defaultUrl,
started: defaultUrl != "", started: defaultUrl != "",
} }
@ -99,7 +97,7 @@ func (manager *BroacastManagerCtx) createPipeline() error {
Str("src", pipelineStr). Str("src", pipelineStr).
Msgf("starting pipeline") Msgf("starting pipeline")
manager.pipeline, err = gst.CreatePipeline(pipelineStr, manager.sampleChannel) manager.pipeline, err = gst.CreatePipeline(pipelineStr)
if err != nil { if err != nil {
return err return err
} }

11
server/internal/capture/gst/gst.go
View File

@ -37,7 +37,7 @@ func init() {
registry = C.gst_registry_get() registry = C.gst_registry_get()
} }
func CreatePipeline(pipelineStr string, sampleChannel chan types.Sample) (*Pipeline, error) { func CreatePipeline(pipelineStr string) (*Pipeline, error) {
id := atomic.AddInt32(&pSerial, 1) id := atomic.AddInt32(&pSerial, 1)
pipelineStrUnsafe := C.CString(pipelineStr) pipelineStrUnsafe := C.CString(pipelineStr)
@ -61,19 +61,20 @@ func CreatePipeline(pipelineStr string, sampleChannel chan types.Sample) (*Pipel
Str("module", "capture"). Str("module", "capture").
Str("submodule", "gstreamer"). Str("submodule", "gstreamer").
Int("pipeline_id", int(id)).Logger(), Int("pipeline_id", int(id)).Logger(),
Src: pipelineStr, Src: pipelineStr,
Ctx: ctx, Ctx: ctx,
Sample: sampleChannel,
} }
pipelines[p.id] = p pipelines[p.id] = p
return p, nil return p, nil
} }
func (p *Pipeline) AttachAppsink(sinkName string) { func (p *Pipeline) AttachAppsink(sinkName string, sampleChannel chan types.Sample) {
sinkNameUnsafe := C.CString(sinkName) sinkNameUnsafe := C.CString(sinkName)
defer C.free(unsafe.Pointer(sinkNameUnsafe)) defer C.free(unsafe.Pointer(sinkNameUnsafe))
p.Sample = sampleChannel
C.gstreamer_pipeline_attach_appsink(p.Ctx, sinkNameUnsafe) C.gstreamer_pipeline_attach_appsink(p.Ctx, sinkNameUnsafe)
} }

16
server/internal/capture/streamsink.go
View File

@ -13,9 +13,9 @@ import (
) )
type StreamSinkManagerCtx struct { type StreamSinkManagerCtx struct {
logger zerolog.Logger logger zerolog.Logger
mu sync.Mutex mu sync.Mutex
sampleChannel chan types.Sample sampleChannel chan types.Sample
codec codec.RTPCodec codec codec.RTPCodec
pipeline *gst.Pipeline pipeline *gst.Pipeline
@ -33,9 +33,9 @@ func streamSinkNew(codec codec.RTPCodec, pipelineFn func() (string, error), vide
Str("video_id", video_id).Logger() Str("video_id", video_id).Logger()
manager := &StreamSinkManagerCtx{ manager := &StreamSinkManagerCtx{
logger: logger, logger: logger,
codec: codec, codec: codec,
pipelineFn: pipelineFn, pipelineFn: pipelineFn,
sampleChannel: make(chan types.Sample), sampleChannel: make(chan types.Sample),
} }
@ -141,7 +141,7 @@ func (manager *StreamSinkManagerCtx) createPipeline() error {
Str("src", pipelineStr). Str("src", pipelineStr).
Msgf("creating pipeline") Msgf("creating pipeline")
manager.pipeline, err = gst.CreatePipeline(pipelineStr, manager.sampleChannel) manager.pipeline, err = gst.CreatePipeline(pipelineStr)
if err != nil { if err != nil {
return err return err
} }
@ -151,7 +151,7 @@ func (manager *StreamSinkManagerCtx) createPipeline() error {
appsinkSubfix = "video" appsinkSubfix = "video"
} }
manager.pipeline.AttachAppsink("appsink" + appsinkSubfix) manager.pipeline.AttachAppsink("appsink"+appsinkSubfix, manager.sampleChannel)
manager.pipeline.Play() manager.pipeline.Play()
return nil return nil