From 6d595e8572663f70c855810af468429d0b77de70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0ediv=C3=BD?= Date: Mon, 1 Feb 2021 23:50:18 +0100 Subject: [PATCH] refactor to v3 + custom codec handler. --- internal/capture/gst/gst.go | 48 +++++--------- internal/capture/manager.go | 9 +-- internal/config/capture.go | 31 +++++---- internal/config/webrtc.go | 1 + internal/types/capture.go | 15 +++-- internal/types/codec/codecs.go | 113 +++++++++++++++++++++++++++++++++ internal/webrtc/manager.go | 87 ++++++++++++------------- 7 files changed, 203 insertions(+), 101 deletions(-) create mode 100644 internal/types/codec/codecs.go diff --git a/internal/capture/gst/gst.go b/internal/capture/gst/gst.go index d41608a4..9527c257 100644 --- a/internal/capture/gst/gst.go +++ b/internal/capture/gst/gst.go @@ -8,19 +8,18 @@ package gst import "C" import ( "fmt" + "time" "sync" "unsafe" - "github.com/pion/webrtc/v2" - "demodesk/neko/internal/types" + "demodesk/neko/internal/types/codec" ) // Pipeline is a wrapper for a GStreamer Pipeline type Pipeline struct { Pipeline *C.GstElement Sample chan types.Sample - ClockRate float32 Src string id int } @@ -30,9 +29,6 @@ var pipelinesLock sync.Mutex var registry *C.GstRegistry const ( - videoClockRate = 90000 - audioClockRate = 48000 - pcmClockRate = 8000 videoSrc = "ximagesrc display-name=%s show-pointer=false use-damage=false ! video/x-raw ! videoconvert ! queue ! " audioSrc = "pulsesrc device=%s ! audio/x-raw,channels=2 ! audioconvert ! " appSink = " ! appsink name=appsink" @@ -55,7 +51,7 @@ func CreateRTMPPipeline(pipelineDevice string, pipelineDisplay string, pipelineS pipelineStr = fmt.Sprintf("flvmux name=mux ! rtmpsink location='%s live=1' %s audio/x-raw,channels=2 ! audioconvert ! voaacenc ! mux. %s x264enc bframes=0 key-int-max=60 byte-stream=true tune=zerolatency speed-preset=veryfast ! mux.", pipelineRTMP, audio, video) } - return CreatePipeline(pipelineStr, 0) + return CreatePipeline(pipelineStr) } // CreateJPEGPipeline creates a GStreamer Pipeline @@ -67,16 +63,15 @@ func CreateJPEGPipeline(pipelineDisplay string, pipelineSrc string, rate string, pipelineStr = fmt.Sprintf("ximagesrc display-name=%s show-pointer=true use-damage=false ! videoconvert ! videoscale ! videorate ! video/x-raw,framerate=%s ! jpegenc quality=%s" + appSink, pipelineDisplay, rate, quality) } - return CreatePipeline(pipelineStr, 0) + return CreatePipeline(pipelineStr) } // CreateAppPipeline creates a GStreamer Pipeline -func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc string) (*Pipeline, error) { - var clockRate float32 +func CreateAppPipeline(codecRTP codec.RTP, pipelineDevice string, pipelineSrc string) (*Pipeline, error) { var pipelineStr string - switch codecName { - case webrtc.VP8: + switch codecRTP.Name { + case codec.VP8: // https://gstreamer.freedesktop.org/documentation/vpx/vp8enc.html?gi-language=c // gstreamer1.0-plugins-good // vp8enc error-resilient=partitions keyframe-max-dist=10 auto-alt-ref=true cpu-used=5 deadline=1 @@ -84,9 +79,8 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri return nil, err } - clockRate = videoClockRate pipelineStr = fmt.Sprintf(videoSrc + "vp8enc cpu-used=8 threads=2 deadline=1 error-resilient=partitions keyframe-max-dist=10 auto-alt-ref=true" + appSink, pipelineDevice) - case webrtc.VP9: + case codec.VP9: // https://gstreamer.freedesktop.org/documentation/vpx/vp9enc.html?gi-language=c // gstreamer1.0-plugins-good // vp9enc @@ -94,9 +88,8 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri return nil, err } - clockRate = videoClockRate pipelineStr = fmt.Sprintf(videoSrc + "vp9enc" + appSink, pipelineDevice) - case webrtc.H264: + case codec.H264: if err := CheckPlugins([]string{"ximagesrc"}); err != nil { return nil, err } @@ -106,7 +99,6 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri // gstreamer1.0-plugins-bad // openh264enc multi-thread=4 complexity=high bitrate=3072000 max-bitrate=4096000 if err = CheckPlugins([]string{"openh264"}); err == nil { - clockRate = videoClockRate pipelineStr = fmt.Sprintf(videoSrc + "openh264enc multi-thread=4 complexity=high bitrate=3072000 max-bitrate=4096000 ! video/x-h264,stream-format=byte-stream" + appSink, pipelineDevice) break } @@ -115,13 +107,12 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri // gstreamer1.0-plugins-ugly // video/x-raw,format=I420 ! x264enc bframes=0 key-int-max=60 byte-stream=true tune=zerolatency speed-preset=veryfast ! video/x-h264,stream-format=byte-stream if err = CheckPlugins([]string{"x264"}); err == nil { - clockRate = videoClockRate pipelineStr = fmt.Sprintf(videoSrc + "video/x-raw,format=I420 ! x264enc bframes=0 key-int-max=60 byte-stream=true tune=zerolatency speed-preset=veryfast ! video/x-h264,stream-format=byte-stream" + appSink, pipelineDevice) break } return nil, err - case webrtc.Opus: + case codec.Opus: // https://gstreamer.freedesktop.org/documentation/opus/opusenc.html // gstreamer1.0-plugins-base // opusenc @@ -129,9 +120,8 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri return nil, err } - clockRate = audioClockRate pipelineStr = fmt.Sprintf(audioSrc + "opusenc" + appSink, pipelineDevice) - case webrtc.G722: + case codec.G722: // https://gstreamer.freedesktop.org/documentation/libav/avenc_g722.html?gi-language=c // gstreamer1.0-libav // avenc_g722 @@ -139,9 +129,8 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri return nil, err } - clockRate = audioClockRate pipelineStr = fmt.Sprintf(audioSrc + "avenc_g722" + appSink, pipelineDevice) - case webrtc.PCMU: + case codec.PCMU: // https://gstreamer.freedesktop.org/documentation/mulaw/mulawenc.html?gi-language=c // gstreamer1.0-plugins-good // audio/x-raw, rate=8000 ! mulawenc @@ -149,9 +138,8 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri return nil, err } - clockRate = pcmClockRate pipelineStr = fmt.Sprintf(audioSrc + "audio/x-raw, rate=8000 ! mulawenc" + appSink, pipelineDevice) - case webrtc.PCMA: + case codec.PCMA: // https://gstreamer.freedesktop.org/documentation/alaw/alawenc.html?gi-language=c // gstreamer1.0-plugins-good // audio/x-raw, rate=8000 ! alawenc @@ -159,21 +147,20 @@ func CreateAppPipeline(codecName string, pipelineDevice string, pipelineSrc stri return nil, err } - clockRate = pcmClockRate pipelineStr = fmt.Sprintf(audioSrc + "audio/x-raw, rate=8000 ! alawenc" + appSink, pipelineDevice) default: - return nil, fmt.Errorf("unknown codec %s", codecName) + return nil, fmt.Errorf("unknown codec %s", codecRTP.Name) } if pipelineSrc != "" { pipelineStr = fmt.Sprintf(pipelineSrc + appSink, pipelineDevice) } - return CreatePipeline(pipelineStr, clockRate) + return CreatePipeline(pipelineStr) } // CreatePipeline creates a GStreamer Pipeline -func CreatePipeline(pipelineStr string, clockRate float32) (*Pipeline, error) { +func CreatePipeline(pipelineStr string) (*Pipeline, error) { pipelineStrUnsafe := C.CString(pipelineStr) defer C.free(unsafe.Pointer(pipelineStrUnsafe)) @@ -193,7 +180,6 @@ func CreatePipeline(pipelineStr string, clockRate float32) (*Pipeline, error) { p := &Pipeline{ Pipeline: gstPipeline, Sample: make(chan types.Sample), - ClockRate: clockRate, Src: pipelineStr, id: len(pipelines), } @@ -243,7 +229,7 @@ func goHandlePipelineBuffer(buffer unsafe.Pointer, bufferLen C.int, duration C.i if ok { pipeline.Sample <- types.Sample{ Data: C.GoBytes(buffer, bufferLen), - Samples: uint32(pipeline.ClockRate * (float32(duration) / 1e9)), + Duration: time.Duration(duration), } } else { fmt.Printf("discarding buffer, no pipeline with id %d", int(pipelineID)) diff --git a/internal/capture/manager.go b/internal/capture/manager.go index 46770ac2..2116b394 100644 --- a/internal/capture/manager.go +++ b/internal/capture/manager.go @@ -8,6 +8,7 @@ import ( "github.com/rs/zerolog/log" "demodesk/neko/internal/types" + "demodesk/neko/internal/types/codec" "demodesk/neko/internal/config" "demodesk/neko/internal/capture/gst" ) @@ -125,11 +126,11 @@ func (manager *CaptureManagerCtx) Screencast() types.ScreencastManager { return manager.screencast } -func (manager *CaptureManagerCtx) VideoCodec() string { +func (manager *CaptureManagerCtx) VideoCodec() codec.RTP { return manager.config.VideoCodec } -func (manager *CaptureManagerCtx) AudioCodec() string { +func (manager *CaptureManagerCtx) AudioCodec() codec.RTP { return manager.config.AudioCodec } @@ -178,7 +179,7 @@ func (manager *CaptureManagerCtx) createVideoPipeline() { var err error manager.logger.Info(). - Str("video_codec", manager.config.VideoCodec). + Str("video_codec", manager.config.VideoCodec.Name). Str("video_display", manager.config.Display). Str("video_params", manager.config.VideoParams). Msgf("creating video pipeline") @@ -212,7 +213,7 @@ func (manager *CaptureManagerCtx) createAudioPipeline() { var err error manager.logger.Info(). - Str("audio_codec", manager.config.AudioCodec). + Str("audio_codec", manager.config.AudioCodec.Name). Str("audio_display", manager.config.Device). Str("audio_params", manager.config.AudioParams). Msgf("creating audio pipeline") diff --git a/internal/config/capture.go b/internal/config/capture.go index 397367b4..fb072d02 100644 --- a/internal/config/capture.go +++ b/internal/config/capture.go @@ -1,17 +1,18 @@ package config import ( - "github.com/pion/webrtc/v2" "github.com/spf13/cobra" "github.com/spf13/viper" + + "demodesk/neko/internal/types/codec" ) type Capture struct { Display string Device string - AudioCodec string + AudioCodec codec.RTP AudioParams string - VideoCodec string + VideoCodec codec.RTP VideoParams string BroadcastPipeline string @@ -110,24 +111,30 @@ func (Capture) Init(cmd *cobra.Command) error { } func (s *Capture) Set() { - videoCodec := webrtc.VP8 + var videoCodec codec.RTP if viper.GetBool("vp8") { - videoCodec = webrtc.VP8 + videoCodec = codec.New(codec.VP8) } else if viper.GetBool("vp9") { - videoCodec = webrtc.VP9 + videoCodec = codec.New(codec.VP9) } else if viper.GetBool("h264") { - videoCodec = webrtc.H264 + videoCodec = codec.New(codec.H264) + } else { + // default + videoCodec = codec.New(codec.VP8) } - audioCodec := webrtc.Opus + var audioCodec codec.RTP if viper.GetBool("opus") { - audioCodec = webrtc.Opus + audioCodec = codec.New(codec.Opus) } else if viper.GetBool("g722") { - audioCodec = webrtc.G722 + audioCodec = codec.New(codec.G722) } else if viper.GetBool("pcmu") { - audioCodec = webrtc.PCMU + audioCodec = codec.New(codec.PCMU) } else if viper.GetBool("pcma") { - audioCodec = webrtc.PCMA + audioCodec = codec.New(codec.PCMA) + } else { + // default + audioCodec = codec.New(codec.Opus) } s.Device = viper.GetString("device") diff --git a/internal/config/webrtc.go b/internal/config/webrtc.go index 13cbf185..faa23768 100644 --- a/internal/config/webrtc.go +++ b/internal/config/webrtc.go @@ -6,6 +6,7 @@ import ( "github.com/spf13/cobra" "github.com/spf13/viper" + "demodesk/neko/internal/utils" ) diff --git a/internal/types/capture.go b/internal/types/capture.go index 6481dfe4..7af5d01f 100644 --- a/internal/types/capture.go +++ b/internal/types/capture.go @@ -1,9 +1,12 @@ package types -type Sample struct { - Data []byte - Samples uint32 -} +import ( + "github.com/pion/webrtc/v3/pkg/media" + + "demodesk/neko/internal/types/codec" +) + +type Sample media.Sample type BroadcastManager interface { Start(url string) error @@ -25,8 +28,8 @@ type CaptureManager interface { Broadcast() BroadcastManager Screencast() ScreencastManager - VideoCodec() string - AudioCodec() string + VideoCodec() codec.RTP + AudioCodec() codec.RTP OnVideoFrame(listener func(sample Sample)) OnAudioFrame(listener func(sample Sample)) diff --git a/internal/types/codec/codecs.go b/internal/types/codec/codecs.go new file mode 100644 index 00000000..4ffc4db7 --- /dev/null +++ b/internal/types/codec/codecs.go @@ -0,0 +1,113 @@ +package codec + +import "github.com/pion/webrtc/v3" + +const ( + VP8 = "vp8" + VP9 = "vp9" + H264 = "h264" + Opus = "opus" + G722 = "g722" + PCMU = "pcmu" + PCMA = "pcma" +) + +type RTP struct { + Name string + PayloadType webrtc.PayloadType + Type webrtc.RTPCodecType + Capability webrtc.RTPCodecCapability +} + +func New(codecType string) RTP { + codec := RTP{} + + switch codecType { + case "vp8": + codec.Name = "vp8" + codec.PayloadType = 96 + codec.Type = webrtc.RTPCodecTypeVideo + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeVP8, + ClockRate: 90000, + Channels: 0, + SDPFmtpLine: "", + RTCPFeedback: nil, + } + case "vp9": + codec.Name = "vp9" + codec.PayloadType = 98 + codec.Type = webrtc.RTPCodecTypeVideo + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeVP9, + ClockRate: 90000, + Channels: 0, + SDPFmtpLine: "profile-id=0", + RTCPFeedback: nil, + } + case "h264": + codec.Name = "h264" + codec.PayloadType = 102 + codec.Type = webrtc.RTPCodecTypeVideo + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeH264, + ClockRate: 90000, + Channels: 0, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f", + RTCPFeedback: nil, + } + case "opus": + codec.Name = "opus" + codec.PayloadType = 111 + codec.Type = webrtc.RTPCodecTypeAudio + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeOpus, + ClockRate: 48000, + Channels: 2, + SDPFmtpLine: "", + RTCPFeedback: nil, + } + case "g722": + codec.Name = "g722" + codec.PayloadType = 9 + codec.Type = webrtc.RTPCodecTypeAudio + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeG722, + ClockRate: 8000, + Channels: 0, + SDPFmtpLine: "", + RTCPFeedback: nil, + } + case "pcmu": + codec.Name = "pcmu" + codec.PayloadType = 0 + codec.Type = webrtc.RTPCodecTypeAudio + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMU, + ClockRate: 8000, + Channels: 0, + SDPFmtpLine: "", + RTCPFeedback: nil, + } + case "pcma": + codec.Name = "pcma" + codec.PayloadType = 8 + codec.Type = webrtc.RTPCodecTypeAudio + codec.Capability = webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypePCMA, + ClockRate: 8000, + Channels: 0, + SDPFmtpLine: "", + RTCPFeedback: nil, + } + } + + return codec +} + +func (codec *RTP) Register(engine *webrtc.MediaEngine) error { + return engine.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: codec.Capability, + PayloadType: codec.PayloadType, + }, codec.Type) +} diff --git a/internal/webrtc/manager.go b/internal/webrtc/manager.go index e0346383..fdfd491a 100644 --- a/internal/webrtc/manager.go +++ b/internal/webrtc/manager.go @@ -3,15 +3,15 @@ package webrtc import ( "fmt" "io" - "math/rand" "strings" - "github.com/pion/webrtc/v2" - "github.com/pion/webrtc/v2/pkg/media" + "github.com/pion/webrtc/v3" + "github.com/pion/webrtc/v3/pkg/media" "github.com/rs/zerolog" "github.com/rs/zerolog/log" "demodesk/neko/internal/types" + "demodesk/neko/internal/types/codec" "demodesk/neko/internal/config" ) @@ -26,10 +26,10 @@ func New(desktop types.DesktopManager, capture types.CaptureManager, config *con type WebRTCManagerCtx struct { logger zerolog.Logger - videoTrack *webrtc.Track - audioTrack *webrtc.Track - videoCodec *webrtc.RTPCodec - audioCodec *webrtc.RTPCodec + videoTrack *webrtc.TrackLocalStaticSample + audioTrack *webrtc.TrackLocalStaticSample + videoCodec codec.RTP + audioCodec codec.RTP desktop types.DesktopManager capture types.CaptureManager config *config.WebRTC @@ -38,7 +38,9 @@ type WebRTCManagerCtx struct { func (manager *WebRTCManagerCtx) Start() { var err error - manager.audioTrack, manager.audioCodec, err = manager.createTrack(manager.capture.AudioCodec()) + // create audio track + manager.audioCodec = manager.capture.AudioCodec() + manager.audioTrack, err = webrtc.NewTrackLocalStaticSample(manager.audioCodec.Capability, "audio", "stream") if err != nil { manager.logger.Panic().Err(err).Msg("unable to create audio track") } @@ -49,7 +51,9 @@ func (manager *WebRTCManagerCtx) Start() { } }) - manager.videoTrack, manager.videoCodec, err = manager.createTrack(manager.capture.VideoCodec()) + // create video track + manager.videoCodec = manager.capture.VideoCodec() + manager.videoTrack, err = webrtc.NewTrackLocalStaticSample(manager.videoCodec.Capability, "video", "stream") if err != nil { manager.logger.Panic().Err(err).Msg("unable to create video track") } @@ -93,6 +97,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session) (string, bool configuration = &webrtc.Configuration{ SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback, } + settings.SetLite(true) } @@ -104,10 +109,15 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session) (string, bool settings.SetNAT1To1IPs(manager.config.NAT1To1IPs, webrtc.ICECandidateTypeHost) // Create MediaEngine based off sdp - engine := webrtc.MediaEngine{} + engine := &webrtc.MediaEngine{} - engine.RegisterCodec(manager.audioCodec) - engine.RegisterCodec(manager.videoCodec) + if err := manager.videoCodec.Register(engine); err != nil { + return "", manager.config.ICELite, manager.config.ICEServers, err + } + + if err := manager.audioCodec.Register(engine); err != nil { + return "", manager.config.ICELite, manager.config.ICEServers, err + } // Create API with MediaEngine and SettingEngine api := webrtc.NewAPI(webrtc.WithMediaEngine(engine), webrtc.WithSettingEngine(settings)) @@ -118,6 +128,11 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session) (string, bool return "", manager.config.ICELite, manager.config.ICEServers, err } + _, err = connection.CreateDataChannel("data", nil) + if err != nil { + return "", manager.config.ICELite, manager.config.ICEServers, err + } + if _, err = connection.AddTransceiverFromTrack(manager.videoTrack, webrtc.RtpTransceiverInit{ Direction: webrtc.RTPTransceiverDirectionSendonly, }); err != nil { @@ -130,27 +145,32 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session) (string, bool return "", manager.config.ICELite, manager.config.ICEServers, err } - description, err := connection.CreateOffer(nil) + offer, err := connection.CreateOffer(nil) if err != nil { return "", manager.config.ICELite, manager.config.ICEServers, err } - connection.OnDataChannel(func(d *webrtc.DataChannel) { - d.OnMessage(func(msg webrtc.DataChannelMessage) { + connection.OnDataChannel(func(channel *webrtc.DataChannel) { + channel.OnMessage(func(message webrtc.DataChannelMessage) { if !session.IsHost() { return } - if err = manager.handle(msg); err != nil { + if err = manager.handle(message); err != nil { manager.logger.Warn().Err(err).Msg("data handle failed") } }) }) - if err := connection.SetLocalDescription(description); err != nil { + // TODO: Refactor, send request to client. + gatherComplete := webrtc.GatheringCompletePromise(connection) + + if err := connection.SetLocalDescription(offer); err != nil { return "", manager.config.ICELite, manager.config.ICEServers, err } + <-gatherComplete + connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) { switch state { case webrtc.PeerConnectionStateConnected: @@ -170,40 +190,11 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session) (string, bool session.SetWebRTCPeer(&WebRTCPeerCtx{ api: api, - engine: &engine, + engine: engine, settings: &settings, connection: connection, configuration: configuration, }) - return description.SDP, manager.config.ICELite, manager.config.ICEServers, nil -} - -func (m *WebRTCManagerCtx) createTrack(codecName string) (*webrtc.Track, *webrtc.RTPCodec, error) { - var codec *webrtc.RTPCodec - switch codecName { - case webrtc.VP8: - codec = webrtc.NewRTPVP8Codec(webrtc.DefaultPayloadTypeVP8, 90000) - case webrtc.VP9: - codec = webrtc.NewRTPVP9Codec(webrtc.DefaultPayloadTypeVP9, 90000) - case webrtc.H264: - codec = webrtc.NewRTPH264Codec(webrtc.DefaultPayloadTypeH264, 90000) - case webrtc.Opus: - codec = webrtc.NewRTPOpusCodec(webrtc.DefaultPayloadTypeOpus, 48000) - case webrtc.G722: - codec = webrtc.NewRTPG722Codec(webrtc.DefaultPayloadTypeG722, 8000) - case webrtc.PCMU: - codec = webrtc.NewRTPPCMUCodec(webrtc.DefaultPayloadTypePCMU, 8000) - case webrtc.PCMA: - codec = webrtc.NewRTPPCMACodec(webrtc.DefaultPayloadTypePCMA, 8000) - default: - return nil, nil, fmt.Errorf("unknown codec %s", codecName) - } - - track, err := webrtc.NewTrack(codec.PayloadType, rand.Uint32(), "stream", "stream", codec) - if err != nil { - return nil, nil, err - } - - return track, codec, nil + return connection.LocalDescription().SDP, manager.config.ICELite, manager.config.ICEServers, nil }