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

remove wait timer from goroutine

Browse Source
This commit is contained in:
mbattista 2023-01-29 18:49:40 +01:00 committed by Miroslav Šedivý
parent 79a1c41938
commit 628c6a1b77
4 changed files with 11 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -13,6 +13,7 @@ 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
@ -31,6 +32,7 @@ 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 != "",
} }
@ -97,7 +99,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.pipeline, err = gst.CreatePipeline(pipelineStr, manager.sampleChannel)
if err != nil { if err != nil {
return err return err
} }

4
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) (*Pipeline, error) { func CreatePipeline(pipelineStr string, sampleChannel chan types.Sample) (*Pipeline, error) {
id := atomic.AddInt32(&pSerial, 1) id := atomic.AddInt32(&pSerial, 1)
pipelineStrUnsafe := C.CString(pipelineStr) pipelineStrUnsafe := C.CString(pipelineStr)
@ -63,7 +63,7 @@ func CreatePipeline(pipelineStr string) (*Pipeline, error) {
Int("pipeline_id", int(id)).Logger(), Int("pipeline_id", int(id)).Logger(),
Src: pipelineStr, Src: pipelineStr,
Ctx: ctx, Ctx: ctx,
Sample: make(chan types.Sample), Sample: sampleChannel,
} }
pipelines[p.id] = p pipelines[p.id] = p

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

@ -15,6 +15,7 @@ import (
type StreamSinkManagerCtx struct { type StreamSinkManagerCtx struct {
logger zerolog.Logger logger zerolog.Logger
mu sync.Mutex mu sync.Mutex
sampleChannel chan types.Sample
codec codec.RTPCodec codec codec.RTPCodec
pipeline *gst.Pipeline pipeline *gst.Pipeline
@ -35,6 +36,7 @@ func streamSinkNew(codec codec.RTPCodec, pipelineFn func() (string, error), vide
logger: logger, logger: logger,
codec: codec, codec: codec,
pipelineFn: pipelineFn, pipelineFn: pipelineFn,
sampleChannel: make(chan types.Sample),
} }
return manager return manager
@ -139,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.pipeline, err = gst.CreatePipeline(pipelineStr, manager.sampleChannel)
if err != nil { if err != nil {
return err return err
} }
@ -169,9 +171,5 @@ func (manager *StreamSinkManagerCtx) destroyPipeline() {
} }
func (manager *StreamSinkManagerCtx) GetSampleChannel() chan types.Sample { func (manager *StreamSinkManagerCtx) GetSampleChannel() chan types.Sample {
if manager.pipeline != nil { return manager.sampleChannel
return manager.pipeline.Sample
}
return nil
} }

16
server/internal/webrtc/webrtc.go
View File

@ -57,16 +57,10 @@ func (manager *WebRTCManager) Start() {
go func() { go func() {
for { for {
if manager.capture.Audio().GetSampleChannel() == nil {
// Pipeline not yet initialized
time.Sleep(50 * time.Millisecond)
continue
}
sample, ok := <-manager.capture.Audio().GetSampleChannel() sample, ok := <-manager.capture.Audio().GetSampleChannel()
if !ok { if !ok {
manager.logger.Debug().Msg("audio capture channel is closed") manager.logger.Debug().Msg("audio capture channel is closed")
continue // TOOD: Create this goroutine when creating the pipeline. continue
} }
err := manager.audioTrack.WriteSample(media.Sample(sample)) err := manager.audioTrack.WriteSample(media.Sample(sample))
@ -88,16 +82,10 @@ func (manager *WebRTCManager) Start() {
go func() { go func() {
for { for {
if manager.capture.Video().GetSampleChannel() == nil {
// Pipeline not yet initialized
time.Sleep(50 * time.Millisecond)
continue
}
sample, ok := <-manager.capture.Video().GetSampleChannel() sample, ok := <-manager.capture.Video().GetSampleChannel()
if !ok { if !ok {
manager.logger.Debug().Msg("video capture channel is closed") manager.logger.Debug().Msg("video capture channel is closed")
continue // TOOD: Create this goroutine when creating the pipeline. continue
} }
err := manager.videoTrack.WriteSample(media.Sample(sample)) err := manager.videoTrack.WriteSample(media.Sample(sample))