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

Capture stream: Add- and RemoveListener.

Browse Source
This commit is contained in:
Miroslav Šedivý 2021-02-06 12:52:02 +01:00
parent 6756114e30
commit f05889bd56
3 changed files with 31 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
internal/capture/stream.go
View File

@ -3,8 +3,8 @@ package capture
import ( import (
"fmt" "fmt"
"sync" "sync"
"reflect"
"github.com/kataras/go-events"
"github.com/rs/zerolog" "github.com/rs/zerolog"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
@ -20,7 +20,8 @@ type StreamManagerCtx struct {
pipelineStr string pipelineStr string
pipeline *gst.Pipeline pipeline *gst.Pipeline
sample chan types.Sample sample chan types.Sample
emmiter events.EventEmmiter listeners map[uintptr]func(sample types.Sample)
emitMu sync.Mutex
emitUpdate chan bool emitUpdate chan bool
emitStop chan bool emitStop chan bool
started bool started bool
@ -31,7 +32,7 @@ func streamNew(codec codec.RTPCodec, pipelineStr string) *StreamManagerCtx {
logger: log.With().Str("module", "capture").Str("submodule", "stream").Logger(), logger: log.With().Str("module", "capture").Str("submodule", "stream").Logger(),
codec: codec, codec: codec,
pipelineStr: pipelineStr, pipelineStr: pipelineStr,
emmiter: events.New(), listeners: map[uintptr]func(sample types.Sample){},
emitUpdate: make(chan bool), emitUpdate: make(chan bool),
emitStop: make(chan bool), emitStop: make(chan bool),
started: false, started: false,
@ -48,7 +49,11 @@ func streamNew(codec codec.RTPCodec, pipelineStr string) *StreamManagerCtx {
case <-manager.emitUpdate: case <-manager.emitUpdate:
manager.logger.Debug().Msg("update emitting samples") manager.logger.Debug().Msg("update emitting samples")
case sample := <-manager.sample: case sample := <-manager.sample:
manager.emmiter.Emit("sample", sample) manager.emitMu.Lock()
for _, emit := range manager.listeners {
emit(sample)
}
manager.emitMu.Unlock()
} }
} }
}() }()
@ -67,10 +72,20 @@ func (manager *StreamManagerCtx) Codec() codec.RTPCodec {
return manager.codec return manager.codec
} }
func (manager *StreamManagerCtx) OnSample(listener func(sample types.Sample)) { func (manager *StreamManagerCtx) AddListener(listener func(sample types.Sample)) {
manager.emmiter.On("sample", func(payload ...interface{}) { manager.emitMu.Lock()
listener(payload[0].(types.Sample)) defer manager.emitMu.Unlock()
})
ptr := reflect.ValueOf(listener).Pointer()
manager.listeners[ptr] = listener
}
func (manager *StreamManagerCtx) RemoveListener(listener func(sample types.Sample)) {
manager.emitMu.Lock()
defer manager.emitMu.Unlock()
ptr := reflect.ValueOf(listener).Pointer()
delete(manager.listeners, ptr)
} }
func (manager *StreamManagerCtx) Start() error { func (manager *StreamManagerCtx) Start() error {

3
internal/types/capture.go
View File

@ -23,7 +23,8 @@ type ScreencastManager interface {
type StreamManager interface { type StreamManager interface {
Codec() codec.RTPCodec Codec() codec.RTPCodec
OnSample(listener func(sample Sample)) AddListener(listener func(sample Sample))
RemoveListener(listener func(sample Sample))
Start() error Start() error
Stop() Stop()

8
internal/webrtc/manager.go
View File

@ -48,7 +48,7 @@ func (manager *WebRTCManagerCtx) Start() {
manager.logger.Panic().Err(err).Msg("unable to create audio track") manager.logger.Panic().Err(err).Msg("unable to create audio track")
} }
audio.OnSample(func(sample types.Sample) { audio.AddListener(func(sample types.Sample) {
if err := manager.audioTrack.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe { if err := manager.audioTrack.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe {
manager.logger.Warn().Err(err).Msg("audio pipeline failed to write") manager.logger.Warn().Err(err).Msg("audio pipeline failed to write")
} }
@ -70,7 +70,7 @@ func (manager *WebRTCManagerCtx) Start() {
manager.logger.Panic().Err(err).Str("videoID", videoID).Msg("unable to create video track") manager.logger.Panic().Err(err).Str("videoID", videoID).Msg("unable to create video track")
} }
video.OnSample(func(sample types.Sample) { video.AddListener(func(sample types.Sample) {
if err := track.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe { if err := track.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe {
manager.logger.Warn().Err(err).Str("videoID", videoID).Msg("vide pipeline failed to write") manager.logger.Warn().Err(err).Str("videoID", videoID).Msg("vide pipeline failed to write")
} }
@ -178,6 +178,10 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session) (*webrtc.Sess
} }
} }
connection.OnNegotiationNeeded(func() {
logger.Warn().Msg("!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!negotiation needed!")
})
connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
switch state { switch state {
case webrtc.PeerConnectionStateConnected: case webrtc.PeerConnectionStateConnected: