diff --git a/machinery/src/webrtc/main.go b/machinery/src/webrtc/main.go index aece224..557257d 100644 --- a/machinery/src/webrtc/main.go +++ b/machinery/src/webrtc/main.go @@ -16,6 +16,9 @@ import ( "github.com/kerberos-io/agent/machinery/src/packets" mqtt "github.com/eclipse/paho.mqtt.golang" + "github.com/pion/interceptor" + "github.com/pion/interceptor/pkg/intervalpli" + "github.com/pion/webrtc/v4" pionWebRTC "github.com/pion/webrtc/v4" pionMedia "github.com/pion/webrtc/v4/pkg/media" ) @@ -81,6 +84,19 @@ func RegisterCandidates(key string, candidate models.ReceiveHDCandidatesPayload) CandidatesMutex.Unlock() } +func RegisterDefaultInterceptors(mediaEngine *pionWebRTC.MediaEngine, interceptorRegistry *interceptor.Registry) error { + if err := pionWebRTC.ConfigureNack(mediaEngine, interceptorRegistry); err != nil { + return err + } + if err := pionWebRTC.ConfigureRTCPReports(interceptorRegistry); err != nil { + return err + } + if err := pionWebRTC.ConfigureSimulcastExtensionHeaders(mediaEngine); err != nil { + return err + } + return nil +} + func InitializeWebRTCConnection(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, videoTrack *pionWebRTC.TrackLocalStaticSample, audioTrack *pionWebRTC.TrackLocalStaticSample, handshake models.RequestHDStreamPayload) { config := configuration.Config @@ -114,7 +130,31 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong registering codecs for media engine: " + err.Error()) } - api := pionWebRTC.NewAPI(pionWebRTC.WithMediaEngine(mediaEngine)) + // Create a InterceptorRegistry. This is the user configurable RTP/RTCP Pipeline. + // This provides NACKs, RTCP Reports and other features. If you use `webrtc.NewPeerConnection` + // this is enabled by default. If you are manually managing You MUST create a InterceptorRegistry + // for each PeerConnection. + interceptorRegistry := &interceptor.Registry{} + + // Use the default set of Interceptors + if err := webrtc.RegisterDefaultInterceptors(mediaEngine, interceptorRegistry); err != nil { + panic(err) + } + + // Register a intervalpli factory + // This interceptor sends a PLI every 3 seconds. A PLI causes a video keyframe to be generated by the sender. + // This makes our video seekable and more error resilent, but at a cost of lower picture quality and higher bitrates + // A real world application should process incoming RTCP packets from viewers and forward them to senders + intervalPliFactory, err := intervalpli.NewReceiverInterceptor() + if err != nil { + panic(err) + } + interceptorRegistry.Add(intervalPliFactory) + + api := pionWebRTC.NewAPI( + pionWebRTC.WithMediaEngine(mediaEngine), + pionWebRTC.WithInterceptorRegistry(interceptorRegistry), + ) policy := pionWebRTC.ICETransportPolicyAll if config.ForceTurn == "true" { @@ -139,13 +179,36 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati if err == nil && peerConnection != nil { - if _, err = peerConnection.AddTrack(videoTrack); err != nil { + var videoSender *pionWebRTC.RTPSender = nil + if videoSender, err = peerConnection.AddTrack(videoTrack); err != nil { log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding video track: " + err.Error()) } + // Read incoming RTCP packets + // Before these packets are returned they are processed by interceptors. For things + // like NACK this needs to be called. + go func() { + rtcpBuf := make([]byte, 1500) + for { + if _, _, rtcpErr := videoSender.Read(rtcpBuf); rtcpErr != nil { + return + } + } + }() - if _, err = peerConnection.AddTrack(audioTrack); err != nil { + var audioSender *pionWebRTC.RTPSender = nil + if audioSender, err = peerConnection.AddTrack(audioTrack); err != nil { log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding audio track: " + err.Error()) - } + } // Read incoming RTCP packets + // Before these packets are returned they are processed by interceptors. For things + // like NACK this needs to be called. + go func() { + rtcpBuf := make([]byte, 1500) + for { + if _, _, rtcpErr := audioSender.Read(rtcpBuf); rtcpErr != nil { + return + } + } + }() peerConnection.OnConnectionStateChange(func(connectionState pionWebRTC.PeerConnectionState) { if connectionState == pionWebRTC.PeerConnectionStateDisconnected || connectionState == pionWebRTC.PeerConnectionStateClosed {