dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-17 18:32:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use StreamStateChange with a single list (#235)

Browse Source 
* Use StreamStateChange with a single list

Use enum for state change indication on associated track.

* update protocol version
This commit is contained in:
Raja Subramanian
2021-12-04 11:57:23 +05:30
committed by GitHub
parent c41384cd09
commit 45690bc301
6 changed files with 54 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/participant.go
+15 -16
View File
@@ -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,
},
})
}
pkg/rtc/transport.go
+2 -2
View File
@@ -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) {
pkg/service/wire_gen.go
+2 -1
View File
@@ -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
pkg/sfu/streamallocator.go
+32 -23
View File
@@ -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
}
//------------------------------------------------