Send adaptive stream param in join (#626)

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)

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

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
 }

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
 }
 

pkg/rtc/types/interfaces.go

@@ -87,6 +87,7 @@ type LocalParticipant interface {
 	Participant
 
 	GetLogger() logger.Logger
+	GetAdaptiveStream() bool
 
 	ProtocolVersion() ProtocolVersion
 

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()

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)

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
 }