From 45690bc3018852b54bdd05596682e30f545bdc8e Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 4 Dec 2021 11:57:23 +0530 Subject: [PATCH] Use StreamStateChange with a single list (#235) * Use StreamStateChange with a single list Use enum for state change indication on associated track. * update protocol version --- go.mod | 2 +- go.sum | 4 +-- pkg/rtc/participant.go | 31 +++++++++++---------- pkg/rtc/transport.go | 4 +-- pkg/service/wire_gen.go | 3 ++- pkg/sfu/streamallocator.go | 55 ++++++++++++++++++++++---------------- 6 files changed, 54 insertions(+), 45 deletions(-) diff --git a/go.mod b/go.mod index 09861c979..ae5b674ac 100644 --- a/go.mod +++ b/go.mod @@ -15,7 +15,7 @@ require ( github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b - github.com/livekit/protocol v0.10.3 + github.com/livekit/protocol v0.10.4 github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 diff --git a/go.sum b/go.sum index 5d12996f3..e38ad5b72 100644 --- a/go.sum +++ b/go.sum @@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v0.10.3 h1:Jofe/vmPc9DB1o7T7nhMZxjIhuThlQBLU5F9Ju0WQjg= -github.com/livekit/protocol v0.10.3/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= +github.com/livekit/protocol v0.10.4 h1:pn4Du5o8EfDpjHPhkjez8qHRYzNNy6ZP2c0T87cQKc0= +github.com/livekit/protocol v0.10.4/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 6c747439e..0f5141502 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -182,7 +182,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { p.subscriber.OnOffer(p.onOffer) - p.subscriber.OnStreamedTracksChange(p.onStreamedTracksChange) + p.subscriber.OnStreamStateChange(p.onStreamStateChange) return p, nil } @@ -1319,28 +1319,27 @@ func (p *ParticipantImpl) configureReceiverDTX() { } } -func (p *ParticipantImpl) onStreamedTracksChange(update *sfu.StreamedTracksUpdate) error { - if len(update.Paused) == 0 && len(update.Resumed) == 0 { +func (p *ParticipantImpl) onStreamStateChange(update *sfu.StreamStateUpdate) error { + if len(update.StreamStates) == 0 { return nil } - streamedTracksUpdate := &livekit.StreamedTracksUpdate{} - for _, streamedTrack := range update.Paused { - streamedTracksUpdate.Paused = append(streamedTracksUpdate.Paused, &livekit.StreamedTrack{ - ParticipantSid: streamedTrack.ParticipantSid, - TrackSid: streamedTrack.TrackSid, - }) - } - for _, streamedTrack := range update.Resumed { - streamedTracksUpdate.Resumed = append(streamedTracksUpdate.Resumed, &livekit.StreamedTrack{ - ParticipantSid: streamedTrack.ParticipantSid, - TrackSid: streamedTrack.TrackSid, + streamStateUpdate := &livekit.StreamStateUpdate{} + for _, streamStateInfo := range update.StreamStates { + state := livekit.StreamState_ACTIVE + if streamStateInfo.State == sfu.StreamStatePaused { + state = livekit.StreamState_PAUSED + } + streamStateUpdate.StreamStates = append(streamStateUpdate.StreamStates, &livekit.StreamStateInfo{ + ParticipantSid: streamStateInfo.ParticipantSid, + TrackSid: streamStateInfo.TrackSid, + State: state, }) } return p.writeMessage(&livekit.SignalResponse{ - Message: &livekit.SignalResponse_StreamedTracksUpdate{ - StreamedTracksUpdate: streamedTracksUpdate, + Message: &livekit.SignalResponse_StreamStateUpdate{ + StreamStateUpdate: streamStateUpdate, }, }) } diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 585b53873..d1987ea89 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -256,12 +256,12 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { return nil } -func (t *PCTransport) OnStreamedTracksChange(f func(update *sfu.StreamedTracksUpdate) error) { +func (t *PCTransport) OnStreamStateChange(f func(update *sfu.StreamStateUpdate) error) { if t.streamAllocator == nil { return } - t.streamAllocator.OnStreamedTracksChange(f) + t.streamAllocator.OnStreamStateChange(f) } func (t *PCTransport) AddTrack(subTrack types.SubscribedTrack) { diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index fafb4ac46..29265178c 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -1,7 +1,8 @@ // Code generated by Wire. DO NOT EDIT. //go:generate go run github.com/google/wire/cmd/wire -//+build !wireinject +//go:build !wireinject +// +build !wireinject package service diff --git a/pkg/sfu/streamallocator.go b/pkg/sfu/streamallocator.go index bfd7551c1..76ba8fdca 100644 --- a/pkg/sfu/streamallocator.go +++ b/pkg/sfu/streamallocator.go @@ -200,7 +200,7 @@ type StreamAllocator struct { participantID string logger logger.Logger - onStreamedTracksChange func(update *StreamedTracksUpdate) error + onStreamStateChange func(update *StreamStateUpdate) error trackingSSRC uint32 committedChannelCapacity int64 @@ -266,8 +266,8 @@ func (s *StreamAllocator) Stop() { close(s.eventCh) } -func (s *StreamAllocator) OnStreamedTracksChange(f func(update *StreamedTracksUpdate) error) { - s.onStreamedTracksChange = f +func (s *StreamAllocator) OnStreamStateChange(f func(update *StreamStateUpdate) error) { + s.onStreamStateChange = f } func (s *StreamAllocator) AddTrack(downTrack *DownTrack) { @@ -698,7 +698,7 @@ func (s *StreamAllocator) maybeCommitEstimate() (isDecreasing bool) { func (s *StreamAllocator) allocateTrack(track *Track) { // if not deficient, free pass allocate track if s.state == StateStable { - update := NewStreamedTracksUpdate() + update := NewStreamStateUpdate() result := track.Allocate(ChannelCapacityInfinity) update.HandleStreamingChange(result.change, track) s.maybeSendUpdate(update) @@ -727,7 +727,7 @@ func (s *StreamAllocator) allocateTrack(track *Track) { // track may be downgrading due to mute or reduction in subscribed layers // and actually giving back some bits. // - update := NewStreamedTracksUpdate() + update := NewStreamStateUpdate() result := track.TryAllocate(lpExpectedBps) @@ -765,7 +765,7 @@ func (s *StreamAllocator) unallocateTrack(track *Track) { return } - update := NewStreamedTracksUpdate() + update := NewStreamStateUpdate() unallocatedBps := track.BandwidthRequested() if unallocatedBps > 0 { @@ -777,7 +777,7 @@ func (s *StreamAllocator) unallocateTrack(track *Track) { s.adjustState() } -func (s *StreamAllocator) tryAllocateTracks(tracks []*Track, additionalBps int64, update *StreamedTracksUpdate) int64 { +func (s *StreamAllocator) tryAllocateTracks(tracks []*Track, additionalBps int64, update *StreamStateUpdate) int64 { for _, t := range tracks { if !t.IsDeficient() { continue @@ -820,7 +820,7 @@ func (s *StreamAllocator) allocateAllTracks() { // // Ask down tracks adjust their forwarded layers. // - update := NewStreamedTracksUpdate() + update := NewStreamStateUpdate() availableChannelCapacity := s.committedChannelCapacity for _, track := range s.videoTracksSorted { @@ -859,14 +859,14 @@ func (s *StreamAllocator) allocateAllTracks() { s.adjustState() } -func (s *StreamAllocator) maybeSendUpdate(update *StreamedTracksUpdate) { +func (s *StreamAllocator) maybeSendUpdate(update *StreamStateUpdate) { if update.Empty() { return } - s.logger.Debugw("streamed tracks changed", "participant", s.participantID, "paused", update.Paused, "resumed", update.Resumed) - if s.onStreamedTracksChange != nil { - err := s.onStreamedTracksChange(update) + s.logger.Debugw("streamed tracks changed", "participant", s.participantID, "update", update) + if s.onStreamStateChange != nil { + err := s.onStreamStateChange(update) if err != nil { s.logger.Errorw("could not send streamed tracks update", err, "participant", s.participantID) } @@ -1006,37 +1006,46 @@ func (s *StreamAllocator) resetGratuitousProbe() { //------------------------------------------------ -type StreamedTrack struct { +type StreamState int + +const ( + StreamStateActive StreamState = iota + StreamStatePaused +) + +type StreamStateInfo struct { ParticipantSid string TrackSid string + State StreamState } -type StreamedTracksUpdate struct { - Paused []*StreamedTrack - Resumed []*StreamedTrack +type StreamStateUpdate struct { + StreamStates []*StreamStateInfo } -func NewStreamedTracksUpdate() *StreamedTracksUpdate { - return &StreamedTracksUpdate{} +func NewStreamStateUpdate() *StreamStateUpdate { + return &StreamStateUpdate{} } -func (s *StreamedTracksUpdate) HandleStreamingChange(change VideoStreamingChange, track *Track) { +func (s *StreamStateUpdate) HandleStreamingChange(change VideoStreamingChange, track *Track) { switch change { case VideoStreamingChangePausing: - s.Paused = append(s.Paused, &StreamedTrack{ + s.StreamStates = append(s.StreamStates, &StreamStateInfo{ ParticipantSid: track.PeerID(), TrackSid: track.ID(), + State: StreamStatePaused, }) case VideoStreamingChangeResuming: - s.Resumed = append(s.Resumed, &StreamedTrack{ + s.StreamStates = append(s.StreamStates, &StreamStateInfo{ ParticipantSid: track.PeerID(), TrackSid: track.ID(), + State: StreamStateActive, }) } } -func (s *StreamedTracksUpdate) Empty() bool { - return len(s.Paused) == 0 && len(s.Resumed) == 0 +func (s *StreamStateUpdate) Empty() bool { + return len(s.StreamStates) == 0 } //------------------------------------------------