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

stream listener use channel dispatcher.

Browse Source
This commit is contained in:
Miroslav Šedivý 2021-09-29 00:22:51 +02:00
parent 18fbdb2ff8
commit 77d2ef9aaf
3 changed files with 25 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
internal/capture/stream.go
View File

@ -4,6 +4,7 @@ import (
"errors"
"reflect"
"sync"
"time"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
@ -13,12 +14,14 @@ import (
"demodesk/neko/internal/types/codec"
)
const newListenerTimeout = 500 * time.Millisecond
type StreamManagerCtx struct {
logger zerolog.Logger
mu sync.Mutex
wg sync.WaitGroup
codec codec.RTPCodec
codec codec.RTPCodec
pipeline *gst.Pipeline
pipelineMu sync.Mutex
pipelineStr func() string
@ -92,9 +95,9 @@ func (manager *StreamManagerCtx) Codec() codec.RTPCodec {
return manager.codec
}
func (manager *StreamManagerCtx) NewListener(listener *func(sample types.Sample)) (addListener func(), err error) {
func (manager *StreamManagerCtx) NewListener(listener *func(sample types.Sample)) (dispatcher chan interface{}, err error) {
if listener == nil {
return addListener, errors.New("listener cannot be nil")
return dispatcher, errors.New("listener cannot be nil")
}
manager.mu.Lock()
@ -104,13 +107,22 @@ func (manager *StreamManagerCtx) NewListener(listener *func(sample types.Sample)
if manager.listenersCount == 1 {
err := manager.createPipeline()
if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) {
return addListener, err
return dispatcher, err
}
manager.logger.Info().Msgf("first listener, starting")
}
return func() {
dispatcher = make(chan interface{}, 1)
go func() {
select {
case <-time.After(newListenerTimeout):
manager.logger.Warn().Msgf("add listener channel was not called, timeouted")
break
case <-dispatcher:
break
}
ptr := reflect.ValueOf(listener).Pointer()
manager.listenersMu.Lock()
@ -118,7 +130,9 @@ func (manager *StreamManagerCtx) NewListener(listener *func(sample types.Sample)
manager.listenersMu.Unlock()
manager.logger.Debug().Interface("ptr", ptr).Msgf("adding listener")
}, nil
}()
return dispatcher, nil
}
func (manager *StreamManagerCtx) RemoveListener(listener *func(sample types.Sample)) {

5
internal/types/capture.go
View File

@ -35,8 +35,9 @@ type ScreencastManager interface {
type StreamManager interface {
Codec() codec.RTPCodec
// starts pipeline if was not running before and returns register function
NewListener(listener *func(sample Sample)) (addListener func(), err error)
// starts pipeline if was not running before
// and returns dispatcher channel
NewListener(listener *func(sample Sample)) (dispatcher chan interface{}, err error)
// stops pipeline if it was last listener
RemoveListener(listener *func(sample Sample))

4
internal/webrtc/peertrack.go
View File

@ -52,7 +52,7 @@ func (peer *PeerTrack) SetStream(stream types.StreamManager) error {
defer peer.streamMu.Unlock()
// prepare new listener
addListener, err := stream.NewListener(&peer.listener)
dispatcher, err := stream.NewListener(&peer.listener)
if err != nil {
return err
}
@ -63,7 +63,7 @@ func (peer *PeerTrack) SetStream(stream types.StreamManager) error {
}
// add new listener
addListener()
close(dispatcher)
peer.stream = stream
return nil