From cf627d8bbed6fde700cb66f62d50e444839416b6 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 19 Apr 2022 16:45:35 +0530 Subject: [PATCH] Send adaptive stream param in join (#626) --- pkg/routing/interfaces.go | 15 +++-- pkg/rtc/mediatracksubscriptions.go | 1 + pkg/rtc/participant.go | 5 ++ pkg/rtc/subscribedtrack.go | 5 +- pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 65 +++++++++++++++++++ pkg/service/roommanager.go | 1 + pkg/service/rtcservice.go | 4 ++ 8 files changed, 89 insertions(+), 8 deletions(-) diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index f51811505..9b3e131e8 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -29,13 +29,14 @@ type MessageSource interface { } type ParticipantInit struct { - Identity livekit.ParticipantIdentity - Name livekit.ParticipantName - Reconnect bool - AutoSubscribe bool - Client *livekit.ClientInfo - Grants *auth.ClaimGrants - Region string + Identity livekit.ParticipantIdentity + Name livekit.ParticipantName + Reconnect bool + AutoSubscribe bool + Client *livekit.ClientInfo + Grants *auth.ClaimGrants + Region string + AdaptiveStream bool } type NewParticipantCallback func(ctx context.Context, roomName livekit.RoomName, pi ParticipantInit, requestSource MessageSource, responseSink MessageSink) diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index d3cf67672..aeb34be04 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -147,6 +147,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code SubscriberID: subscriberID, MediaTrack: t.params.MediaTrack, DownTrack: downTrack, + AdaptiveStream: sub.GetAdaptiveStream(), }) var transceiver *webrtc.RTPTransceiver diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 452badaa9..81d1b2b40 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -65,6 +65,7 @@ type ParticipantParams struct { ClientConf *livekit.ClientConfiguration Region string Migration bool + AdaptiveStream bool } type ParticipantImpl struct { @@ -293,6 +294,10 @@ func (p *ParticipantImpl) GetLogger() logger.Logger { return p.params.Logger } +func (p *ParticipantImpl) GetAdaptiveStream() bool { + return p.params.AdaptiveStream +} + func (p *ParticipantImpl) ID() livekit.ParticipantID { return p.params.SID } diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index b9f013fd9..b9bc5d5b1 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -23,6 +23,7 @@ type SubscribedTrackParams struct { SubscriberID livekit.ParticipantID MediaTrack types.MediaTrack DownTrack *sfu.DownTrack + AdaptiveStream bool } type SubscribedTrack struct { @@ -42,7 +43,9 @@ func NewSubscribedTrack(params SubscribedTrackParams) *SubscribedTrack { debouncer: debounce.New(subscriptionDebounceInterval), } - s.params.DownTrack.SetMaxSpatialLayer(SpatialLayerForQuality(livekit.VideoQuality_HIGH)) + if !s.params.AdaptiveStream { + s.params.DownTrack.SetMaxSpatialLayer(SpatialLayerForQuality(livekit.VideoQuality_HIGH)) + } return s } diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 8bd5531ac..6ea99b931 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -87,6 +87,7 @@ type LocalParticipant interface { Participant GetLogger() logger.Logger + GetAdaptiveStream() bool ProtocolVersion() ProtocolVersion diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 0c566a43f..1a2a475fe 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -121,6 +121,16 @@ type FakeLocalParticipant struct { debugInfoReturnsOnCall map[int]struct { result1 map[string]interface{} } + GetAdaptiveStreamStub func() bool + getAdaptiveStreamMutex sync.RWMutex + getAdaptiveStreamArgsForCall []struct { + } + getAdaptiveStreamReturns struct { + result1 bool + } + getAdaptiveStreamReturnsOnCall map[int]struct { + result1 bool + } GetAudioLevelStub func() (uint8, bool) getAudioLevelMutex sync.RWMutex getAudioLevelArgsForCall []struct { @@ -1190,6 +1200,59 @@ func (fake *FakeLocalParticipant) DebugInfoReturnsOnCall(i int, result1 map[stri }{result1} } +func (fake *FakeLocalParticipant) GetAdaptiveStream() bool { + fake.getAdaptiveStreamMutex.Lock() + ret, specificReturn := fake.getAdaptiveStreamReturnsOnCall[len(fake.getAdaptiveStreamArgsForCall)] + fake.getAdaptiveStreamArgsForCall = append(fake.getAdaptiveStreamArgsForCall, struct { + }{}) + stub := fake.GetAdaptiveStreamStub + fakeReturns := fake.getAdaptiveStreamReturns + fake.recordInvocation("GetAdaptiveStream", []interface{}{}) + fake.getAdaptiveStreamMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetAdaptiveStreamCallCount() int { + fake.getAdaptiveStreamMutex.RLock() + defer fake.getAdaptiveStreamMutex.RUnlock() + return len(fake.getAdaptiveStreamArgsForCall) +} + +func (fake *FakeLocalParticipant) GetAdaptiveStreamCalls(stub func() bool) { + fake.getAdaptiveStreamMutex.Lock() + defer fake.getAdaptiveStreamMutex.Unlock() + fake.GetAdaptiveStreamStub = stub +} + +func (fake *FakeLocalParticipant) GetAdaptiveStreamReturns(result1 bool) { + fake.getAdaptiveStreamMutex.Lock() + defer fake.getAdaptiveStreamMutex.Unlock() + fake.GetAdaptiveStreamStub = nil + fake.getAdaptiveStreamReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) GetAdaptiveStreamReturnsOnCall(i int, result1 bool) { + fake.getAdaptiveStreamMutex.Lock() + defer fake.getAdaptiveStreamMutex.Unlock() + fake.GetAdaptiveStreamStub = nil + if fake.getAdaptiveStreamReturnsOnCall == nil { + fake.getAdaptiveStreamReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.getAdaptiveStreamReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) GetAudioLevel() (uint8, bool) { fake.getAudioLevelMutex.Lock() ret, specificReturn := fake.getAudioLevelReturnsOnCall[len(fake.getAudioLevelArgsForCall)] @@ -3917,6 +3980,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.connectedAtMutex.RUnlock() fake.debugInfoMutex.RLock() defer fake.debugInfoMutex.RUnlock() + fake.getAdaptiveStreamMutex.RLock() + defer fake.getAdaptiveStreamMutex.RUnlock() fake.getAudioLevelMutex.RLock() defer fake.getAudioLevelMutex.RUnlock() fake.getConnectionQualityMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 1a685aa77..f7152dcda 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -256,6 +256,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam Logger: pLogger, ClientConf: clientConf, Region: pi.Region, + AdaptiveStream: pi.AdaptiveStream, }) if err != nil { logger.Errorw("could not create participant", err) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 0d09dd942..ff2b5a94e 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -92,6 +92,7 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic reconnectParam := r.FormValue("reconnect") autoSubParam := r.FormValue("auto_subscribe") publishParam := r.FormValue("publish") + adaptiveStreamParam := r.FormValue("adaptive_stream") if onlyName != "" { roomName = onlyName @@ -131,6 +132,9 @@ func (s *RTCService) validate(r *http.Request) (livekit.RoomName, routing.Partic if autoSubParam != "" { pi.AutoSubscribe = boolValue(autoSubParam) } + if adaptiveStreamParam != "" { + pi.AdaptiveStream = boolValue(adaptiveStreamParam) + } return roomName, pi, http.StatusOK, nil }