diff --git a/pkg/config/config.go b/pkg/config/config.go index 8b0bd9d53..bb57ca48c 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -62,6 +62,9 @@ type RTCConfig struct { // Throttle periods for pli/fir rtcp packets PLIThrottle PLIThrottleConfig `yaml:"pli_throttle"` + + // Which side runs bandwidth estimation + UseSendSideBWE bool `yaml:"send_side_bandwidth_estimation"` } type PLIThrottleConfig struct { diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index c781287ab..d87c07bc9 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -5,6 +5,7 @@ import ( "net" "github.com/pion/ice/v2" + "github.com/pion/sdp/v3" "github.com/pion/webrtc/v3" "github.com/livekit/livekit-server/pkg/config" @@ -15,6 +16,7 @@ import ( const ( minUDPBufferSize = 5_000_000 defaultUDPBufferSize = 16_777_216 + frameMarking = "urn:ietf:params:rtp-hdrext:framemarking" ) type WebRTCConfig struct { @@ -25,6 +27,8 @@ type WebRTCConfig struct { UDPMux ice.UDPMux UDPMuxConn *net.UDPConn TCPMuxListener *net.TCPListener + Publisher DirectionConfig + Subscriber DirectionConfig } type ReceiverConfig struct { @@ -32,6 +36,21 @@ type ReceiverConfig struct { maxBitrate uint64 } +type RTPHeaderExtensionConfig struct { + Audio []string + Video []string +} + +type RTCPFeedbackConfig struct { + Audio []webrtc.RTCPFeedback + Video []webrtc.RTCPFeedback +} + +type DirectionConfig struct { + RTPHeaderExtension RTPHeaderExtensionConfig + RTCPFeedback RTCPFeedbackConfig +} + // number of packets to buffer up const readBufferSize = 50 @@ -112,6 +131,50 @@ func NewWebRTCConfig(conf *config.Config, externalIP string) (*WebRTCConfig, err } s.SetNetworkTypes(networkTypes) + // publisher configuration + publisherConfig := DirectionConfig{ + RTPHeaderExtension: RTPHeaderExtensionConfig{ + Audio: []string{ + sdp.SDESMidURI, + sdp.SDESRTPStreamIDURI, + sdp.AudioLevelURI, + }, + Video: []string{ + sdp.SDESMidURI, + sdp.SDESRTPStreamIDURI, + sdp.TransportCCURI, + frameMarking, + }, + }, + RTCPFeedback: RTCPFeedbackConfig{ + Video: []webrtc.RTCPFeedback{ + {Type: webrtc.TypeRTCPFBGoogREMB}, + {Type: webrtc.TypeRTCPFBTransportCC}, + {Type: webrtc.TypeRTCPFBCCM, Parameter: "fir"}, + {Type: webrtc.TypeRTCPFBNACK}, + {Type: webrtc.TypeRTCPFBNACK, Parameter: "pli"}, + }, + }, + } + + // subscriber configuration + subscriberConfig := DirectionConfig{ + RTCPFeedback: RTCPFeedbackConfig{ + Video: []webrtc.RTCPFeedback{ + {Type: webrtc.TypeRTCPFBCCM, Parameter: "fir"}, + {Type: webrtc.TypeRTCPFBNACK}, + {Type: webrtc.TypeRTCPFBNACK, Parameter: "pli"}, + }, + }, + } + if rtcConf.UseSendSideBWE { + subscriberConfig.RTPHeaderExtension.Video = append(subscriberConfig.RTPHeaderExtension.Video, sdp.TransportCCURI) + subscriberConfig.RTCPFeedback.Video = append(subscriberConfig.RTCPFeedback.Video, webrtc.RTCPFeedback{Type: webrtc.TypeRTCPFBTransportCC}) + } else { + subscriberConfig.RTPHeaderExtension.Video = append(subscriberConfig.RTPHeaderExtension.Video, sdp.ABSSendTimeURI) + subscriberConfig.RTCPFeedback.Video = append(subscriberConfig.RTCPFeedback.Video, webrtc.RTCPFeedback{Type: webrtc.TypeRTCPFBGoogREMB}) + } + return &WebRTCConfig{ Configuration: c, SettingEngine: s, @@ -122,6 +185,8 @@ func NewWebRTCConfig(conf *config.Config, externalIP string) (*WebRTCConfig, err UDPMux: udpMux, UDPMuxConn: udpMuxConn, TCPMuxListener: tcpListener, + Publisher: publisherConfig, + Subscriber: subscriberConfig, }, nil } diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index 4f9aa0b6a..8da740c44 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -4,16 +4,11 @@ import ( "strings" "github.com/livekit/protocol/livekit" - "github.com/pion/sdp/v3" "github.com/pion/webrtc/v3" ) -const ( - frameMarking = "urn:ietf:params:rtp-hdrext:framemarking" -) - -func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec) error { - opusCodec := webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, SDPFmtpLine: "minptime=10;useinbandfec=1", RTCPFeedback: nil} +func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec, rtcpFeedback RTCPFeedbackConfig) error { + opusCodec := webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, SDPFmtpLine: "minptime=10;useinbandfec=1", RTCPFeedback: rtcpFeedback.Audio} if isCodecEnabled(codecs, opusCodec) { if err := me.RegisterCodec(webrtc.RTPCodecParameters{ RTPCodecCapability: opusCodec, @@ -23,34 +18,29 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec) error { } } - videoRTCPFeedback := []webrtc.RTCPFeedback{ - {Type: webrtc.TypeRTCPFBGoogREMB, Parameter: ""}, - {Type: webrtc.TypeRTCPFBCCM, Parameter: "fir"}, - {Type: webrtc.TypeRTCPFBNACK, Parameter: ""}, - {Type: webrtc.TypeRTCPFBNACK, Parameter: "pli"}} for _, codec := range []webrtc.RTPCodecParameters{ { - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: videoRTCPFeedback}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: rtcpFeedback.Video}, PayloadType: 96, }, { - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=0", RTCPFeedback: videoRTCPFeedback}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=0", RTCPFeedback: rtcpFeedback.Video}, PayloadType: 98, }, { - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=1", RTCPFeedback: videoRTCPFeedback}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=1", RTCPFeedback: rtcpFeedback.Video}, PayloadType: 100, }, { - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", RTCPFeedback: rtcpFeedback.Video}, PayloadType: 125, }, { - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f", RTCPFeedback: rtcpFeedback.Video}, PayloadType: 108, }, { - RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640032", RTCPFeedback: videoRTCPFeedback}, + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640032", RTCPFeedback: rtcpFeedback.Video}, PayloadType: 123, }, } { @@ -63,46 +53,30 @@ func registerCodecs(me *webrtc.MediaEngine, codecs []*livekit.Codec) error { return nil } -func createPubMediaEngine(codecs []*livekit.Codec) (*webrtc.MediaEngine, error) { - me := &webrtc.MediaEngine{} - if err := registerCodecs(me, codecs); err != nil { - return nil, err - } - for _, extension := range []string{ - sdp.SDESMidURI, - sdp.SDESRTPStreamIDURI, - sdp.TransportCCURI, - frameMarking, - } { +func registerHeaderExtensions(me *webrtc.MediaEngine, rtpHeaderExtension RTPHeaderExtensionConfig) error { + for _, extension := range rtpHeaderExtension.Video { if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil { - return nil, err - } - } - for _, extension := range []string{ - sdp.SDESMidURI, - sdp.SDESRTPStreamIDURI, - sdp.AudioLevelURI, - } { - if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeAudio); err != nil { - return nil, err + return err } } - return me, nil + for _, extension := range rtpHeaderExtension.Audio { + if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeAudio); err != nil { + return err + } + } + + return nil } -func createSubMediaEngine(codecs []*livekit.Codec) (*webrtc.MediaEngine, error) { +func createMediaEngine(codecs []*livekit.Codec, config DirectionConfig) (*webrtc.MediaEngine, error) { me := &webrtc.MediaEngine{} - if err := registerCodecs(me, codecs); err != nil { + if err := registerCodecs(me, codecs, config.RTCPFeedback); err != nil { return nil, err } - for _, extension := range []string{ - sdp.ABSSendTimeURI, - } { - if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil { - return nil, err - } + if err := registerHeaderExtensions(me, config.RTPHeaderExtension); err != nil { + return nil, err } return me, nil diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 2d805d5b9..8b6d6ed40 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -25,13 +25,6 @@ import ( "github.com/livekit/livekit-server/pkg/telemetry" ) -var ( - FeedbackTypes = []webrtc.RTCPFeedback{ - {Type: webrtc.TypeRTCPFBGoogREMB}, - {Type: webrtc.TypeRTCPFBNACK}, - {Type: webrtc.TypeRTCPFBNACK, Parameter: "pli"}} -) - const ( lostUpdateDelta = time.Second connectionQualityUpdateInterval = 5 * time.Second @@ -96,6 +89,8 @@ type MediaTrackParams struct { AudioConfig config.AudioConfig Telemetry telemetry.TelemetryService Logger logger.Logger + + SubscriberConfig DirectionConfig } func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack { @@ -220,13 +215,22 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error { // react-native-webrtc still uses stream based APIs and require this streamId = PackStreamID(t.params.ParticipantID, t.ID()) } + receiver := NewWrappedReceiver(t.receiver, t.ID(), streamId) + + var rtcpFeedback []webrtc.RTCPFeedback + switch t.Kind() { + case livekit.TrackType_AUDIO: + rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Audio + case livekit.TrackType_VIDEO: + rtcpFeedback = t.params.SubscriberConfig.RTCPFeedback.Video + } downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{ MimeType: codec.MimeType, ClockRate: codec.ClockRate, Channels: codec.Channels, SDPFmtpLine: codec.SDPFmtpLine, - RTCPFeedback: FeedbackTypes, + RTCPFeedback: rtcpFeedback, }, receiver, t.params.BufferFactory, subscriberID, t.params.ReceiverConfig.PacketBufferSize) if err != nil { return err diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 3cdf2ccca..807e238d7 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -7,6 +7,8 @@ import ( "github.com/bep/debounce" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + "github.com/pion/interceptor" + "github.com/pion/sdp/v3" "github.com/pion/webrtc/v3" "github.com/livekit/livekit-server/pkg/rtc/types" @@ -55,29 +57,78 @@ type TransportParams struct { Logger logger.Logger } +// LK-TODO-SSBWE func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) { func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) { - var me *webrtc.MediaEngine - var err error + var directionConfig DirectionConfig if params.Target == livekit.SignalTarget_PUBLISHER { - me, err = createPubMediaEngine(params.EnabledCodecs) + directionConfig = params.Config.Publisher } else { - me, err = createSubMediaEngine(params.EnabledCodecs) + directionConfig = params.Config.Subscriber } + me, err := createMediaEngine(params.EnabledCodecs, directionConfig) if err != nil { return nil, nil, err } + se := params.Config.SettingEngine se.DisableMediaEngineCopy(true) + ir := &interceptor.Registry{} + if params.Target == livekit.SignalTarget_SUBSCRIBER { + isSendSideBWE := false + for _, ext := range directionConfig.RTPHeaderExtension.Video { + if ext == sdp.TransportCCURI { + isSendSideBWE = true + break + } + } + for _, ext := range directionConfig.RTPHeaderExtension.Audio { + if ext == sdp.TransportCCURI { + isSendSideBWE = true + break + } + } + + if isSendSideBWE { + /* LK-TODO-SSBWE + gf, err := cc.NewInterceptor(func() (cc.BandwidthEstimator, error) { + return gcc.NewSendSideBWE( + gcc.SendSideBWEInitialBitrate(1*1000*1000), + gcc.SendSideBWEPacer(gcc.NewNoOpPacer()), + ) + }) + if err == nil { + gf.OnNewPeerConnection(func(id string, estimator cc.BandwidthEstimator) { + if onBandwidthEstimator != nil { + onBandwidthEstimator(estimator) + } + }) + ir.Add(gf) + + tf, err := twcc.NewHeaderExtensionInterceptor() + if err == nil { + ir.Add(tf) + } + } + */ + } + } api := webrtc.NewAPI( webrtc.WithMediaEngine(me), webrtc.WithSettingEngine(se), + webrtc.WithInterceptorRegistry(ir), ) pc, err := api.NewPeerConnection(params.Config.Configuration) return pc, me, err } func NewPCTransport(params TransportParams) (*PCTransport, error) { + /* LK-TODO-SSBWE + var bwe cc.BandwidthEstimator + pc, me, err := newPeerConnection(params, func(estimator cc.BandwidthEstimator) { + bwe = estimator + }) + */ pc, me, err := newPeerConnection(params) if err != nil { return nil, err @@ -95,6 +146,11 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) { Logger: params.Logger, }) t.streamAllocator.Start() + /* LK-TODO-SSBWE + if bwe != nil { + t.streamAllocator.SetBandwidthEstimator(bwe) + } + */ } t.pc.OnICEGatheringStateChange(func(state webrtc.ICEGathererState) { if state == webrtc.ICEGathererStateComplete { diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index 19233c690..1baacc58a 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -342,6 +342,7 @@ func (u *UptrackManager) MediaTrackReceived(track *webrtc.TrackRemote, rtpReceiv AudioConfig: u.params.AudioConfig, Telemetry: u.params.Telemetry, Logger: u.params.Logger, + SubscriberConfig: u.params.Config.Subscriber, }) mt.OnSubscribedMaxQualityChange(u.onSubscribedMaxQualityChange) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 0e8ce6cb8..347a9b312 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -114,8 +114,9 @@ type DownTrack struct { pktsDropped atomicUint32 // RTCP callbacks - onRTCP func([]rtcp.Packet) - onREMB func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate) + onRTCP func([]rtcp.Packet) + onREMB func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMaximumBitrate) + onTransportCCFeedback func(dt *DownTrack, cc *rtcp.TransportLayerCC) // simulcast layer availability change callback onAvailableLayersChanged func(dt *DownTrack) @@ -497,6 +498,10 @@ func (d *DownTrack) OnREMB(fn func(dt *DownTrack, remb *rtcp.ReceiverEstimatedMa d.onREMB = fn } +func (d *DownTrack) OnTransportCCFeedback(fn func(dt *DownTrack, cc *rtcp.TransportLayerCC)) { + d.onTransportCCFeedback = fn +} + func (d *DownTrack) CurrentMaxLossFraction() uint8 { return d.lossFraction.get() } @@ -851,6 +856,10 @@ func (d *DownTrack) handleRTCP(bytes []byte) { nackedPackets = append(nackedPackets, d.sequencer.getSeqNoPairs(pair.PacketList())...) } go d.retransmitPackets(nackedPackets) + case *rtcp.TransportLayerCC: + if p.MediaSSRC == d.ssrc && d.onTransportCCFeedback != nil { + d.onTransportCCFeedback(d, p) + } } } } diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index 5efc4cf28..76aa6c65e 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -62,6 +62,7 @@ const ( SignalAddTrack Signal = iota SignalRemoveTrack SignalEstimate + SignalTargetBitrate SignalReceiverReport SignalAvailableLayersChange SignalSubscriptionChange @@ -78,6 +79,8 @@ func (s Signal) String() string { return "REMOVE_TRACK" case SignalEstimate: return "ESTIMATE" + case SignalTargetBitrate: + return "TARGET_BITRATE" case SignalReceiverReport: return "RECEIVER_REPORT" case SignalSubscriptionChange: @@ -93,6 +96,16 @@ func (s Signal) String() string { } } +type Event struct { + Signal Signal + DownTrack *DownTrack + Data interface{} +} + +func (e Event) String() string { + return fmt.Sprintf("StreamAllocator:Event{signal: %s, data: %s}", e.Signal, e.Data) +} + type StreamAllocatorParams struct { Logger logger.Logger } @@ -102,7 +115,10 @@ type StreamAllocator struct { onStreamStateChange func(update *StreamStateUpdate) error - trackingSSRC uint32 + rembTrackingSSRC uint32 + + // LK-TODO-SSBWE bwe cc.BandwidthEstimator + committedChannelCapacity int64 lastCommitTime time.Time prevReceivedEstimate int64 @@ -127,16 +143,6 @@ type StreamAllocator struct { runningCh chan struct{} } -type Event struct { - Signal Signal - DownTrack *DownTrack - Data interface{} -} - -func (e Event) String() string { - return fmt.Sprintf("StreamAllocator:Event{signal: %s, data: %s}", e.Signal, e.Data) -} - func NewStreamAllocator(params StreamAllocatorParams) *StreamAllocator { s := &StreamAllocator{ logger: params.Logger, @@ -173,6 +179,15 @@ func (s *StreamAllocator) OnStreamStateChange(f func(update *StreamStateUpdate) s.onStreamStateChange = f } +/* LK-TODO-SSBWE +func (s *StreamAllocator) SetBandwidthEstimator(bwe cc.BandwidthEstimator) { + if bwe != nil { + bwe.OnTargetBitrateChange(s.onTargetBitrateChange) + } + s.bwe = bwe +} +*/ + func (s *StreamAllocator) AddTrack(downTrack *DownTrack, isManaged bool) { s.postEvent(Event{ Signal: SignalAddTrack, @@ -182,6 +197,7 @@ func (s *StreamAllocator) AddTrack(downTrack *DownTrack, isManaged bool) { if downTrack.Kind() == webrtc.RTPCodecTypeVideo { downTrack.OnREMB(s.onREMB) + downTrack.OnTransportCCFeedback(s.onTransportCCFeedback) downTrack.OnAvailableLayersChanged(s.onAvailableLayersChanged) downTrack.OnSubscriptionChanged(s.onSubscriptionChanged) downTrack.OnSubscribedLayersChanged(s.onSubscribedLayersChanged) @@ -215,6 +231,23 @@ func (s *StreamAllocator) onREMB(downTrack *DownTrack, remb *rtcp.ReceiverEstima }) } +// called when a new transport-cc feedback is received +func (s *StreamAllocator) onTransportCCFeedback(downTrack *DownTrack, fb *rtcp.TransportLayerCC) { + /* LK-TODO-SSBWE + if s.bwe != nil { + s.bwe.WriteRTCP([]rtcp.Packet{fb}, nil) + } + */ +} + +// called when target bitrate changes +func (s *StreamAllocator) onTargetBitrateChange(bitrate int) { + s.postEvent(Event{ + Signal: SignalTargetBitrate, + Data: bitrate, + }) +} + // called when a new RTCP Receiver Report is received func (s *StreamAllocator) onReceiverReport(downTrack *DownTrack, rr *rtcp.ReceiverReport) { s.postEvent(Event{ @@ -311,6 +344,8 @@ func (s *StreamAllocator) handleEvent(event *Event) { s.handleSignalRemoveTrack(event) case SignalEstimate: s.handleSignalEstimate(event) + case SignalTargetBitrate: + s.handleSignalTargetBitrate(event) case SignalReceiverReport: s.handleSignalReceiverReport(event) case SignalAvailableLayersChange: @@ -428,7 +463,7 @@ func (s *StreamAllocator) handleSignalEstimate(event *Event) { found := false for _, ssrc := range remb.SSRCs { - if ssrc == s.trackingSSRC { + if ssrc == s.rembTrackingSSRC { found = true break } @@ -442,18 +477,18 @@ func (s *StreamAllocator) handleSignalEstimate(event *Event) { // try to lock to track which is sending this update for _, ssrc := range remb.SSRCs { if ssrc == event.DownTrack.SSRC() { - s.trackingSSRC = event.DownTrack.SSRC() + s.rembTrackingSSRC = event.DownTrack.SSRC() found = true break } } if !found { - s.trackingSSRC = remb.SSRCs[0] + s.rembTrackingSSRC = remb.SSRCs[0] } } - if s.trackingSSRC != event.DownTrack.SSRC() { + if s.rembTrackingSSRC != event.DownTrack.SSRC() { return } @@ -473,6 +508,22 @@ func (s *StreamAllocator) handleSignalEstimate(event *Event) { } } +func (s *StreamAllocator) handleSignalTargetBitrate(event *Event) { + receivedEstimate, _ := event.Data.(int) + s.prevReceivedEstimate = s.receivedEstimate + s.receivedEstimate = int64(receivedEstimate) + if s.prevReceivedEstimate != s.receivedEstimate { + s.logger.Debugw("received new estimate", + "old(bps)", s.prevReceivedEstimate, + "new(bps)", s.receivedEstimate, + ) + } + + if s.maybeCommitEstimate() { + s.allocateAllTracks() + } +} + // LK-TODO-START // Receiver report stats are not used in the current implementation. //