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

rename to StreamSink.

Browse Source
This commit is contained in:
Miroslav Šedivý 2021-12-01 20:30:18 +01:00
parent 065a7a2e84
commit 0a94191a5f
4 changed files with 36 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
internal/capture/manager.go
View File

@ -18,8 +18,8 @@ type CaptureManagerCtx struct {
broadcast *BroacastManagerCtx broadcast *BroacastManagerCtx
screencast *ScreencastManagerCtx screencast *ScreencastManagerCtx
audio *StreamManagerCtx audio *StreamSinkManagerCtx
videos map[string]*StreamManagerCtx videos map[string]*StreamSinkManagerCtx
videoIDs []string videoIDs []string
} }
@ -57,7 +57,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
) )
} }
videos := map[string]*StreamManagerCtx{} videos := map[string]*StreamSinkManagerCtx{}
for video_id, cnf := range config.VideoPipelines { for video_id, cnf := range config.VideoPipelines {
pipelineConf := cnf pipelineConf := cnf
@ -88,7 +88,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
Msg("syntax check for video stream pipeline passed") Msg("syntax check for video stream pipeline passed")
// append to videos // append to videos
videos[video_id] = streamNew(config.VideoCodec, createPipeline, video_id) videos[video_id] = streamSinkNew(config.VideoCodec, createPipeline, video_id)
} }
return &CaptureManagerCtx{ return &CaptureManagerCtx{
@ -97,7 +97,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
broadcast: broadcastNew(broadcastPipeline), broadcast: broadcastNew(broadcastPipeline),
screencast: screencastNew(config.ScreencastEnabled, screencastPipeline), screencast: screencastNew(config.ScreencastEnabled, screencastPipeline),
audio: streamNew(config.AudioCodec, func() string { audio: streamSinkNew(config.AudioCodec, func() string {
if config.AudioPipeline != "" { if config.AudioPipeline != "" {
return config.AudioPipeline return config.AudioPipeline
} }
@ -188,11 +188,11 @@ func (manager *CaptureManagerCtx) Screencast() types.ScreencastManager {
return manager.screencast return manager.screencast
} }
func (manager *CaptureManagerCtx) Audio() types.StreamManager { func (manager *CaptureManagerCtx) Audio() types.StreamSinkManager {
return manager.audio return manager.audio
} }
func (manager *CaptureManagerCtx) Video(videoID string) (types.StreamManager, bool) { func (manager *CaptureManagerCtx) Video(videoID string) (types.StreamSinkManager, bool) {
video, ok := manager.videos[videoID] video, ok := manager.videos[videoID]
return video, ok return video, ok
} }

45
internal/capture/stream.go → internal/capture/streamsink.go
View File

@ -13,9 +13,9 @@ import (
"demodesk/neko/internal/types/codec" "demodesk/neko/internal/types/codec"
) )
var moveListenerMu = sync.Mutex{} var moveSinkListenerMu = sync.Mutex{}
type StreamManagerCtx struct { type StreamSinkManagerCtx struct {
logger zerolog.Logger logger zerolog.Logger
mu sync.Mutex mu sync.Mutex
wg sync.WaitGroup wg sync.WaitGroup
@ -33,13 +33,13 @@ type StreamManagerCtx struct {
listenersMu sync.Mutex listenersMu sync.Mutex
} }
func streamNew(codec codec.RTPCodec, pipelineStr func() string, video_id string) *StreamManagerCtx { func streamSinkNew(codec codec.RTPCodec, pipelineStr func() string, video_id string) *StreamSinkManagerCtx {
logger := log.With(). logger := log.With().
Str("module", "capture"). Str("module", "capture").
Str("submodule", "stream"). Str("submodule", "stream-sink").
Str("video_id", video_id).Logger() Str("video_id", video_id).Logger()
manager := &StreamManagerCtx{ manager := &StreamSinkManagerCtx{
logger: logger, logger: logger,
codec: codec, codec: codec,
pipelineStr: pipelineStr, pipelineStr: pipelineStr,
@ -74,7 +74,7 @@ func streamNew(codec codec.RTPCodec, pipelineStr func() string, video_id string)
return manager return manager
} }
func (manager *StreamManagerCtx) shutdown() { func (manager *StreamSinkManagerCtx) shutdown() {
manager.logger.Info().Msgf("shutdown") manager.logger.Info().Msgf("shutdown")
manager.listenersMu.Lock() manager.listenersMu.Lock()
@ -89,11 +89,11 @@ func (manager *StreamManagerCtx) shutdown() {
manager.wg.Wait() manager.wg.Wait()
} }
func (manager *StreamManagerCtx) Codec() codec.RTPCodec { func (manager *StreamSinkManagerCtx) Codec() codec.RTPCodec {
return manager.codec return manager.codec
} }
func (manager *StreamManagerCtx) start() error { func (manager *StreamSinkManagerCtx) start() error {
if len(manager.listeners) == 0 { if len(manager.listeners) == 0 {
err := manager.createPipeline() err := manager.createPipeline()
if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) { if err != nil && !errors.Is(err, types.ErrCapturePipelineAlreadyExists) {
@ -106,14 +106,14 @@ func (manager *StreamManagerCtx) start() error {
return nil return nil
} }
func (manager *StreamManagerCtx) stop() { func (manager *StreamSinkManagerCtx) stop() {
if len(manager.listeners) == 0 { if len(manager.listeners) == 0 {
manager.destroyPipeline() manager.destroyPipeline()
manager.logger.Info().Msgf("last listener, stopping") manager.logger.Info().Msgf("last listener, stopping")
} }
} }
func (manager *StreamManagerCtx) addListener(listener *func(sample types.Sample)) { func (manager *StreamSinkManagerCtx) addListener(listener *func(sample types.Sample)) {
ptr := reflect.ValueOf(listener).Pointer() ptr := reflect.ValueOf(listener).Pointer()
manager.listenersMu.Lock() manager.listenersMu.Lock()
@ -123,7 +123,7 @@ func (manager *StreamManagerCtx) addListener(listener *func(sample types.Sample)
manager.logger.Debug().Interface("ptr", ptr).Msgf("adding listener") manager.logger.Debug().Interface("ptr", ptr).Msgf("adding listener")
} }
func (manager *StreamManagerCtx) removeListener(listener *func(sample types.Sample)) { func (manager *StreamSinkManagerCtx) removeListener(listener *func(sample types.Sample)) {
ptr := reflect.ValueOf(listener).Pointer() ptr := reflect.ValueOf(listener).Pointer()
manager.listenersMu.Lock() manager.listenersMu.Lock()
@ -133,7 +133,7 @@ func (manager *StreamManagerCtx) removeListener(listener *func(sample types.Samp
manager.logger.Debug().Interface("ptr", ptr).Msgf("removing listener") manager.logger.Debug().Interface("ptr", ptr).Msgf("removing listener")
} }
func (manager *StreamManagerCtx) AddListener(listener *func(sample types.Sample)) error { func (manager *StreamSinkManagerCtx) AddListener(listener *func(sample types.Sample)) error {
manager.mu.Lock() manager.mu.Lock()
defer manager.mu.Unlock() defer manager.mu.Unlock()
@ -152,7 +152,7 @@ func (manager *StreamManagerCtx) AddListener(listener *func(sample types.Sample)
return nil return nil
} }
func (manager *StreamManagerCtx) RemoveListener(listener *func(sample types.Sample)) error { func (manager *StreamSinkManagerCtx) RemoveListener(listener *func(sample types.Sample)) error {
manager.mu.Lock() manager.mu.Lock()
defer manager.mu.Unlock() defer manager.mu.Unlock()
@ -171,12 +171,12 @@ func (manager *StreamManagerCtx) RemoveListener(listener *func(sample types.Samp
// moving listeners between streams ensures, that target pipeline is running // moving listeners between streams ensures, that target pipeline is running
// before listener is added, and stops source pipeline if there are 0 listeners // before listener is added, and stops source pipeline if there are 0 listeners
func (manager *StreamManagerCtx) MoveListenerTo(listener *func(sample types.Sample), stream types.StreamManager) error { func (manager *StreamSinkManagerCtx) MoveListenerTo(listener *func(sample types.Sample), stream types.StreamSinkManager) error {
if listener == nil { if listener == nil {
return errors.New("listener cannot be nil") return errors.New("listener cannot be nil")
} }
targetStream, ok := stream.(*StreamManagerCtx) targetStream, ok := stream.(*StreamSinkManagerCtx)
if !ok { if !ok {
return errors.New("target stream manager does not support moving listeners") return errors.New("target stream manager does not support moving listeners")
} }
@ -186,7 +186,7 @@ func (manager *StreamManagerCtx) MoveListenerTo(listener *func(sample types.Samp
// global mutex, that ensures atomic locking // global mutex, that ensures atomic locking
// lock global mutex // lock global mutex
moveListenerMu.Lock() moveSinkListenerMu.Lock()
// lock source stream // lock source stream
manager.mu.Lock() manager.mu.Lock()
@ -197,7 +197,7 @@ func (manager *StreamManagerCtx) MoveListenerTo(listener *func(sample types.Samp
defer targetStream.mu.Unlock() defer targetStream.mu.Unlock()
// unlock global mutex // unlock global mutex
moveListenerMu.Unlock() moveSinkListenerMu.Unlock()
// start if stopped // start if stopped
if err := targetStream.start(); err != nil { if err := targetStream.start(); err != nil {
@ -214,18 +214,18 @@ func (manager *StreamManagerCtx) MoveListenerTo(listener *func(sample types.Samp
return nil return nil
} }
func (manager *StreamManagerCtx) ListenersCount() int { func (manager *StreamSinkManagerCtx) ListenersCount() int {
manager.listenersMu.Lock() manager.listenersMu.Lock()
defer manager.listenersMu.Unlock() defer manager.listenersMu.Unlock()
return len(manager.listeners) return len(manager.listeners)
} }
func (manager *StreamManagerCtx) Started() bool { func (manager *StreamSinkManagerCtx) Started() bool {
return manager.ListenersCount() > 0 return manager.ListenersCount() > 0
} }
func (manager *StreamManagerCtx) createPipeline() error { func (manager *StreamSinkManagerCtx) createPipeline() error {
manager.pipelineMu.Lock() manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock() defer manager.pipelineMu.Unlock()
@ -235,10 +235,9 @@ func (manager *StreamManagerCtx) createPipeline() error {
var err error var err error
codec := manager.Codec()
pipelineStr := manager.pipelineStr() pipelineStr := manager.pipelineStr()
manager.logger.Info(). manager.logger.Info().
Str("codec", codec.Name). Str("codec", manager.codec.Name).
Str("src", pipelineStr). Str("src", pipelineStr).
Msgf("creating pipeline") Msgf("creating pipeline")
@ -255,7 +254,7 @@ func (manager *StreamManagerCtx) createPipeline() error {
return nil return nil
} }
func (manager *StreamManagerCtx) destroyPipeline() { func (manager *StreamSinkManagerCtx) destroyPipeline() {
manager.pipelineMu.Lock() manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock() defer manager.pipelineMu.Unlock()

8
internal/types/capture.go
View File

@ -32,12 +32,12 @@ type ScreencastManager interface {
Image() ([]byte, error) Image() ([]byte, error)
} }
type StreamManager interface { type StreamSinkManager interface {
Codec() codec.RTPCodec Codec() codec.RTPCodec
AddListener(listener *func(sample Sample)) error AddListener(listener *func(sample Sample)) error
RemoveListener(listener *func(sample Sample)) error RemoveListener(listener *func(sample Sample)) error
MoveListenerTo(listener *func(sample Sample), targetStream StreamManager) error MoveListenerTo(listener *func(sample Sample), targetStream StreamSinkManager) error
ListenersCount() int ListenersCount() int
Started() bool Started() bool
@ -49,8 +49,8 @@ type CaptureManager interface {
Broadcast() BroadcastManager Broadcast() BroadcastManager
Screencast() ScreencastManager Screencast() ScreencastManager
Audio() StreamManager Audio() StreamSinkManager
Video(videoID string) (StreamManager, bool) Video(videoID string) (StreamSinkManager, bool)
VideoIDs() []string VideoIDs() []string
} }

6
internal/webrtc/peerstreamtrack.go
View File

@ -11,7 +11,7 @@ import (
"github.com/rs/zerolog" "github.com/rs/zerolog"
) )
func (manager *WebRTCManagerCtx) newPeerStreamTrack(stream types.StreamManager, logger zerolog.Logger) (*PeerStreamTrack, error) { func (manager *WebRTCManagerCtx) newPeerStreamTrack(stream types.StreamSinkManager, logger zerolog.Logger) (*PeerStreamTrack, error) {
codec := stream.Codec() codec := stream.Codec()
id := codec.Type.String() id := codec.Type.String()
@ -42,11 +42,11 @@ type PeerStreamTrack struct {
track *webrtc.TrackLocalStaticSample track *webrtc.TrackLocalStaticSample
listener func(sample types.Sample) listener func(sample types.Sample)
stream types.StreamManager stream types.StreamSinkManager
streamMu sync.Mutex streamMu sync.Mutex
} }
func (peer *PeerStreamTrack) SetStream(stream types.StreamManager) error { func (peer *PeerStreamTrack) SetStream(stream types.StreamSinkManager) error {
peer.streamMu.Lock() peer.streamMu.Lock()
defer peer.streamMu.Unlock() defer peer.streamMu.Unlock()