Archived
2
0

Compare commits

...
This repository has been archived on 2024-06-24. You can view files and clone it, but cannot push or open issues or pull requests.

1 Commits

Author SHA1 Message Date
Miroslav Šedivý
c873d4d344 screenshare POC. 2023-01-29 21:29:16 +01:00
7 changed files with 331 additions and 1 deletions

View File

@ -5,6 +5,14 @@
<span><b>n</b>.eko</span> <span><b>n</b>.eko</span>
</a> </a>
<ul class="menu"> <ul class="menu">
<li>
<button class="btn" @click="startShareScreen" v-if="!mediaStream">
START SCREEN SHARE
</button>
<button class="btn" @click="stopShareScreen" v-else>
STOP SCREEN SHARE
</button>
</li>
<li> <li>
<i <i
:class="[{ disabled: !admin }, { locked: isLocked('control') }, 'fas', 'fa-mouse']" :class="[{ disabled: !admin }, { locked: isLocked('control') }, 'fas', 'fa-mouse']"
@ -207,5 +215,31 @@
return this.$t(`locks.${resource}.` + (this.isLocked(resource) ? `locked` : `unlocked`)) return this.$t(`locks.${resource}.` + (this.isLocked(resource) ? `locked` : `unlocked`))
} }
//
// Screen Share
//
mediaStream: MediaStream | null = null
mediaRtcpSender: RTCRtpSender | null = null
async startShareScreen() {
// get media stream from user's browser
this.mediaStream = await navigator.mediaDevices
.getDisplayMedia({
video: true,
audio: false,
})
const mediaTrack = this.mediaStream.getVideoTracks()[0];
this.mediaRtcpSender = this.$client.addTrack(mediaTrack, this.mediaStream)
}
async stopShareScreen() {
if (this.mediaStream) {
this.mediaStream.getTracks().forEach(track => track.stop())
this.mediaStream = null
}
if (this.mediaRtcpSender) {
this.$client.removeTrack(this.mediaRtcpSender)
this.mediaRtcpSender = null
}
}
} }
</script> </script>

View File

@ -313,6 +313,22 @@ export abstract class BaseClient extends EventEmitter<BaseEvents> {
this._peer.setRemoteDescription({ type: 'answer', sdp }) this._peer.setRemoteDescription({ type: 'answer', sdp })
} }
public addTrack(track: MediaStreamTrack, ...streams: MediaStream[]): RTCRtpSender {
if (!this._peer) {
throw new Error('peer not connected')
}
return this._peer.addTrack(track, ...streams)
}
public removeTrack(sender: RTCRtpSender) {
if (!this._peer) {
throw new Error('peer not connected')
}
this._peer.removeTrack(sender)
}
private async onMessage(e: MessageEvent) { private async onMessage(e: MessageEvent) {
const { event, ...payload } = JSON.parse(e.data) as WebSocketMessages const { event, ...payload } = JSON.parse(e.data) as WebSocketMessages

View File

@ -2,12 +2,14 @@ package capture
import ( import (
"errors" "errors"
"fmt"
"github.com/rs/zerolog" "github.com/rs/zerolog"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
"m1k1o/neko/internal/config" "m1k1o/neko/internal/config"
"m1k1o/neko/internal/types" "m1k1o/neko/internal/types"
"m1k1o/neko/internal/types/codec"
) )
type CaptureManagerCtx struct { type CaptureManagerCtx struct {
@ -18,6 +20,9 @@ type CaptureManagerCtx struct {
broadcast *BroacastManagerCtx broadcast *BroacastManagerCtx
audio *StreamSinkManagerCtx audio *StreamSinkManagerCtx
video *StreamSinkManagerCtx video *StreamSinkManagerCtx
// source-sinks
screenshare *StreamSrcSinkManagerCtx
} }
func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCtx { func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCtx {
@ -43,6 +48,15 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
} }
return NewVideoPipeline(config.VideoCodec, config.Display, config.VideoPipeline, fps, config.VideoBitrate, config.VideoHWEnc) return NewVideoPipeline(config.VideoCodec, config.Display, config.VideoPipeline, fps, config.VideoBitrate, config.VideoHWEnc)
}, "video"), }, "video"),
// source-sinks
screenshare: streamSrcSinkNew(config.ScreenshareEnabled, map[string]string{
codec.VP8().Name: "appsrc format=time is-live=true do-timestamp=true name=appsrc " +
fmt.Sprintf("! application/x-rtp, payload=%d, encoding-name=VP8-DRAFT-IETF-01 ", codec.VP8().PayloadType) +
"! rtpvp8depay " +
"! appsink name=appsink",
// TODO: Add support for more codecs.
}, "webcam"),
} }
} }
@ -95,6 +109,7 @@ func (manager *CaptureManagerCtx) Start() {
func (manager *CaptureManagerCtx) Shutdown() error { func (manager *CaptureManagerCtx) Shutdown() error {
manager.logger.Info().Msgf("shutdown") manager.logger.Info().Msgf("shutdown")
manager.screenshare.shutdown()
manager.broadcast.shutdown() manager.broadcast.shutdown()
manager.audio.shutdown() manager.audio.shutdown()
@ -114,3 +129,7 @@ func (manager *CaptureManagerCtx) Audio() types.StreamSinkManager {
func (manager *CaptureManagerCtx) Video() types.StreamSinkManager { func (manager *CaptureManagerCtx) Video() types.StreamSinkManager {
return manager.video return manager.video
} }
func (manager *CaptureManagerCtx) Screenshare() types.StreamSrcSinkManager {
return manager.screenshare
}

View File

@ -0,0 +1,137 @@
package capture
import (
"errors"
"sync"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"m1k1o/neko/internal/capture/gst"
"m1k1o/neko/internal/types"
"m1k1o/neko/internal/types/codec"
)
type StreamSrcSinkManagerCtx struct {
logger zerolog.Logger
sampleChannel chan types.Sample
enabled bool
codecPipeline map[string]string // codec -> pipeline
codec codec.RTPCodec
pipeline *gst.Pipeline
pipelineMu sync.Mutex
pipelineStr string
}
func streamSrcSinkNew(enabled bool, codecPipeline map[string]string, video_id string) *StreamSrcSinkManagerCtx {
logger := log.With().
Str("module", "capture").
Str("submodule", "stream-src-sink").
Str("video_id", video_id).Logger()
return &StreamSrcSinkManagerCtx{
logger: logger,
enabled: enabled,
codecPipeline: codecPipeline,
sampleChannel: make(chan types.Sample),
}
}
func (manager *StreamSrcSinkManagerCtx) shutdown() {
manager.logger.Info().Msgf("shutdown")
manager.Stop()
}
func (manager *StreamSrcSinkManagerCtx) Codec() codec.RTPCodec {
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
return manager.codec
}
func (manager *StreamSrcSinkManagerCtx) Start(codec codec.RTPCodec) error {
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
if manager.pipeline != nil {
return types.ErrCapturePipelineAlreadyExists
}
if !manager.enabled {
return errors.New("stream-src-sink not enabled")
}
found := false
for codecName, pipeline := range manager.codecPipeline {
if codecName == codec.Name {
manager.pipelineStr = pipeline
manager.codec = codec
found = true
break
}
}
if !found {
return errors.New("no pipeline found for a codec")
}
var err error
manager.logger.Info().
Str("codec", manager.codec.Name).
Str("src", manager.pipelineStr).
Msgf("creating pipeline")
manager.pipeline, err = gst.CreatePipeline(manager.pipelineStr)
if err != nil {
return err
}
manager.pipeline.AttachAppsrc("appsrc")
manager.pipeline.AttachAppsink("appsink", manager.sampleChannel)
manager.pipeline.Play()
return nil
}
func (manager *StreamSrcSinkManagerCtx) Stop() {
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
if manager.pipeline == nil {
return
}
manager.pipeline.Destroy()
manager.pipeline = nil
manager.logger.Info().
Str("codec", manager.codec.Name).
Str("src", manager.pipelineStr).
Msgf("destroying pipeline")
}
func (manager *StreamSrcSinkManagerCtx) Push(bytes []byte) {
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
if manager.pipeline == nil {
return
}
manager.pipeline.Push(bytes)
}
func (manager *StreamSrcSinkManagerCtx) Started() bool {
manager.pipelineMu.Lock()
defer manager.pipelineMu.Unlock()
return manager.pipeline != nil
}
func (manager *StreamSrcSinkManagerCtx) GetSampleChannel() chan types.Sample {
return manager.sampleChannel
}

View File

@ -27,6 +27,9 @@ type Capture struct {
// broadcast // broadcast
BroadcastPipeline string BroadcastPipeline string
BroadcastUrl string BroadcastUrl string
// screenshare
ScreenshareEnabled bool
} }
func (Capture) Init(cmd *cobra.Command) error { func (Capture) Init(cmd *cobra.Command) error {
@ -151,6 +154,15 @@ func (Capture) Init(cmd *cobra.Command) error {
return err return err
} }
//
// screenshare
//
cmd.PersistentFlags().Bool("screenshare.enabled", true, "enable screenshare")
if err := viper.BindPFlag("screenshare.enabled", cmd.PersistentFlags().Lookup("screenshare.enabled")); err != nil {
return err
}
return nil return nil
} }
@ -230,4 +242,10 @@ func (s *Capture) Set() {
s.BroadcastPipeline = viper.GetString("broadcast_pipeline") s.BroadcastPipeline = viper.GetString("broadcast_pipeline")
s.BroadcastUrl = viper.GetString("broadcast_url") s.BroadcastUrl = viper.GetString("broadcast_url")
//
// screenshare
//
s.ScreenshareEnabled = viper.GetBool("screenshare.enabled")
} }

View File

@ -28,6 +28,17 @@ type StreamSinkManager interface {
GetSampleChannel() chan Sample GetSampleChannel() chan Sample
} }
type StreamSrcSinkManager interface {
Codec() codec.RTPCodec
Start(codec codec.RTPCodec) error
Stop()
Push(bytes []byte)
Started() bool
GetSampleChannel() chan Sample
}
type CaptureManager interface { type CaptureManager interface {
Start() Start()
Shutdown() error Shutdown() error
@ -35,4 +46,5 @@ type CaptureManager interface {
Broadcast() BroadcastManager Broadcast() BroadcastManager
Audio() StreamSinkManager Audio() StreamSinkManager
Video() StreamSinkManager Video() StreamSinkManager
Screenshare() StreamSrcSinkManager
} }

View File

@ -11,6 +11,7 @@ import (
"github.com/pion/ice/v2" "github.com/pion/ice/v2"
"github.com/pion/interceptor" "github.com/pion/interceptor"
"github.com/pion/rtcp"
"github.com/pion/webrtc/v3" "github.com/pion/webrtc/v3"
"github.com/pion/webrtc/v3/pkg/media" "github.com/pion/webrtc/v3/pkg/media"
"github.com/rs/zerolog" "github.com/rs/zerolog"
@ -18,6 +19,7 @@ import (
"m1k1o/neko/internal/config" "m1k1o/neko/internal/config"
"m1k1o/neko/internal/types" "m1k1o/neko/internal/types"
"m1k1o/neko/internal/types/codec"
"m1k1o/neko/internal/webrtc/pionlog" "m1k1o/neko/internal/webrtc/pionlog"
) )
@ -40,6 +42,8 @@ type WebRTCManager struct {
desktop types.DesktopManager desktop types.DesktopManager
config *config.WebRTC config *config.WebRTC
api *webrtc.API api *webrtc.API
screenshareStop *func()
} }
func (manager *WebRTCManager) Start() { func (manager *WebRTCManager) Start() {
@ -82,7 +86,19 @@ func (manager *WebRTCManager) Start() {
go func() { go func() {
for { for {
sample, ok := <-manager.capture.Video().GetSampleChannel() var sample types.Sample
var ok bool
select {
case sample, ok = <-manager.capture.Video().GetSampleChannel():
// if screenshare is active, we need to drop all video samples
// ideally we would stop the video capture meanwhile.
if manager.capture.Screenshare().Started() {
continue
}
case sample, ok = <-manager.capture.Screenshare().GetSampleChannel():
}
if !ok { if !ok {
manager.logger.Debug().Msg("video capture channel is closed") manager.logger.Debug().Msg("video capture channel is closed")
continue continue
@ -305,6 +321,84 @@ func (manager *WebRTCManager) CreatePeer(id string, session types.Session) (type
} }
}) })
connection.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
logger := manager.logger.With().
Str("kind", track.Kind().String()).
Str("mime", track.Codec().RTPCodecCapability.MimeType).
Logger()
logger.Info().Msgf("received new remote track")
// parse codec from remote track
codec, ok := codec.ParseRTC(track.Codec())
if !ok {
logger.Warn().Msg("remote track with unknown codec")
receiver.Stop()
return
}
var srcSinkManager types.StreamSrcSinkManager
stopped := false
stopFn := func() {
if stopped {
return
}
stopped = true
receiver.Stop()
srcSinkManager.Stop()
logger.Info().Msg("remote track stopped")
}
logger.Info().Msgf("found codec %s", codec.Name)
if track.Kind() == webrtc.RTPCodecTypeVideo {
// video -> webcam
srcSinkManager = manager.capture.Screenshare()
defer stopFn()
if manager.screenshareStop != nil {
(*manager.screenshareStop)()
}
manager.screenshareStop = &stopFn
} else {
logger.Warn().Msg("expected only video tracks")
receiver.Stop()
return
}
logger.Info().Msg("starting srcSinkManager")
err := srcSinkManager.Start(codec)
if err != nil {
logger.Err(err).Msg("failed to start pipeline")
return
}
ticker := time.NewTicker(3 * time.Second)
defer ticker.Stop()
go func() {
for range ticker.C {
err := connection.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}})
if err != nil {
logger.Err(err).Msg("remote track rtcp send err")
}
}
}()
buf := make([]byte, 1400)
for {
i, _, err := track.Read(buf)
if err != nil {
logger.Warn().Err(err).Msg("failed read from remote track")
break
}
srcSinkManager.Push(buf[:i])
}
})
if err := session.SetPeer(peer); err != nil { if err := session.SetPeer(peer); err != nil {
return nil, err return nil, err
} }