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

refactor to v3 + custom codec handler.

Browse Source
This commit is contained in:
Miroslav Šedivý 2021-02-01 23:50:18 +01:00
parent 4e79c487e5
commit 6d595e8572
7 changed files with 203 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
internal/capture/gst/gst.go
View File

@ -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))

9
internal/capture/manager.go
View File

@ -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")

31
internal/config/capture.go
View File

@ -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")

1
internal/config/webrtc.go
View File

@ -6,6 +6,7 @@ import (
"github.com/spf13/cobra"
"github.com/spf13/viper"
"demodesk/neko/internal/utils"
)

15
internal/types/capture.go
View File

@ -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))

113
internal/types/codec/codecs.go Normal file
View File

@ -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)
}

87
internal/webrtc/manager.go
View File

@ -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
}