diff --git a/go.mod b/go.mod index ab2bb0b79..e09d9f061 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded - github.com/livekit/protocol v1.39.1-0.20250610020455-4ba773363c7f + github.com/livekit/protocol v1.39.2-0.20250612021623-821e6560dbf7 github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 @@ -34,7 +34,7 @@ require ( github.com/pion/datachannel v1.5.10 github.com/pion/dtls/v3 v3.0.6 github.com/pion/ice/v4 v4.0.10 - github.com/pion/interceptor v0.1.38 + github.com/pion/interceptor v0.1.40 github.com/pion/rtcp v1.2.15 github.com/pion/rtp v1.8.18 github.com/pion/sctp v1.8.39 diff --git a/go.sum b/go.sum index 5e5c14284..200a67e2f 100644 --- a/go.sum +++ b/go.sum @@ -169,8 +169,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded h1:ylZPdnlX1RW9Z15SD4mp87vT2D2shsk0hpLJwSPcq3g= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A= -github.com/livekit/protocol v1.39.1-0.20250610020455-4ba773363c7f h1:/CvQV9cfA6oIro/rEfX8YQSCu82s1PPaAwm94b7z7XE= -github.com/livekit/protocol v1.39.1-0.20250610020455-4ba773363c7f/go.mod h1:6HPISM0bkTXTk9RIaQTCe0IDbomBPz7Jwp+N3w5sqL0= +github.com/livekit/protocol v1.39.2-0.20250612021623-821e6560dbf7 h1:JbsygjpjYid8KhUI3yka4djsigIFxmsB0AR65O0SQJw= +github.com/livekit/protocol v1.39.2-0.20250612021623-821e6560dbf7/go.mod h1:6HPISM0bkTXTk9RIaQTCe0IDbomBPz7Jwp+N3w5sqL0= github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c h1:WwEr0YBejYbKzk8LSaO9h8h0G9MnE7shyDu8yXQWmEc= github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c/go.mod h1:kmD+AZPkWu0MaXIMv57jhNlbiSZZ/Jx4bzlxBDVmJes= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= @@ -238,8 +238,8 @@ github.com/pion/dtls/v3 v3.0.6 h1:7Hkd8WhAJNbRgq9RgdNh1aaWlZlGpYTzdqjy9x9sK2E= github.com/pion/dtls/v3 v3.0.6/go.mod h1:iJxNQ3Uhn1NZWOMWlLxEEHAN5yX7GyPvvKw04v9bzYU= github.com/pion/ice/v4 v4.0.10 h1:P59w1iauC/wPk9PdY8Vjl4fOFL5B+USq1+xbDcN6gT4= github.com/pion/ice/v4 v4.0.10/go.mod h1:y3M18aPhIxLlcO/4dn9X8LzLLSma84cx6emMSu14FGw= -github.com/pion/interceptor v0.1.38 h1:Mgt3XIIq47uR5vcLLahfRucE6tFPjxHak+z5ZZFEzLU= -github.com/pion/interceptor v0.1.38/go.mod h1:HS9X+Ue5LDE6q2C2tuvOuO83XkBdJFgn6MBDtfoJX4Q= +github.com/pion/interceptor v0.1.40 h1:e0BjnPcGpr2CFQgKhrQisBU7V3GXK6wrfYrGYaU6Jq4= +github.com/pion/interceptor v0.1.40/go.mod h1:Z6kqH7M/FYirg3frjGJ21VLSRJGBXB/KqaTIrdqnOic= github.com/pion/logging v0.2.3 h1:gHuf0zpoh1GW67Nr6Gj4cv5Z9ZscU7g/EaoC/Ke/igI= github.com/pion/logging v0.2.3/go.mod h1:z8YfknkquMe1csOrxK5kc+5/ZPAzMxbKLX5aXpbpC90= github.com/pion/mdns/v2 v2.0.7 h1:c9kM8ewCgjslaAmicYMFQIde2H9/lrZpjBkN8VwoVtM= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index a6c26c2a9..a6faf05c8 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -81,6 +81,7 @@ type MediaTrackParams struct { ForwardStats *sfu.ForwardStats OnTrackEverSubscribed func(livekit.TrackID) ShouldRegressCodec func() bool + Rids buffer.VideoLayersRid } func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { @@ -106,6 +107,7 @@ func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { Telemetry: params.Telemetry, Logger: params.Logger, RegressionTargetCodec: t.regressionTargetCodec, + Rids: params.Rids, }, ti) if ti.Type == livekit.TrackType_AUDIO { @@ -263,7 +265,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe t.lock.Lock() var regressCodec bool mimeType := mime.NormalizeMimeType(track.Codec().MimeType) - layer := buffer.RidToSpatialLayer(track.RID(), ti) + layer := buffer.RidToSpatialLayer(track.RID(), ti, t.params.Rids) t.params.Logger.Debugw( "AddReceiver", "rid", track.RID(), @@ -302,6 +304,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe receiver, track, ti, + t.params.Rids, LoggerWithCodecMime(t.params.Logger, mimeType), t.params.OnRTCP, t.params.VideoConfig.StreamTrackerManager, diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 43a7570e6..0e8f1f39b 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -130,6 +130,7 @@ type MediaTrackReceiverParams struct { Telemetry telemetry.TelemetryService Logger logger.Logger RegressionTargetCodec mime.MimeType + Rids buffer.VideoLayersRid } type MediaTrackReceiver struct { @@ -669,7 +670,7 @@ func (t *MediaTrackReceiver) updateTrackInfoOfReceivers() { func (t *MediaTrackReceiver) SetLayerSsrc(mimeType mime.MimeType, rid string, ssrc uint32) { t.lock.Lock() trackInfo := t.TrackInfoClone() - layer := buffer.RidToSpatialLayer(rid, trackInfo) + layer := buffer.RidToSpatialLayer(rid, trackInfo, t.params.Rids) if layer == buffer.InvalidLayerSpatial { // non-simulcast case will not have `rid` layer = 0 diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 6764e22f5..0b2817163 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -27,10 +27,10 @@ import ( "github.com/frostbyte73/core" lru "github.com/hashicorp/golang-lru/v2" "github.com/pion/rtcp" - "github.com/pion/sdp/v3" "github.com/pion/webrtc/v4" "github.com/pkg/errors" "go.uber.org/atomic" + "go.uber.org/zap/zapcore" "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil/pkg/twcc" @@ -39,6 +39,7 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/protocol/observability" "github.com/livekit/protocol/observability/roomobs" + sdpHelper "github.com/livekit/protocol/sdp" "github.com/livekit/protocol/utils" "github.com/livekit/protocol/utils/guid" "github.com/livekit/protocol/utils/pointer" @@ -72,17 +73,35 @@ const ( PingTimeoutSeconds = 15 ) +// ------------------------------------------------- + type pendingTrackInfo struct { trackInfos []*livekit.TrackInfo + sdpRids buffer.VideoLayersRid migrated bool createdAt time.Time // indicates if this track is queued for publishing to avoid a track has been published - // before the previous track is unpublished(closed) because client is allowed to neogtiate + // before the previous track is unpublished(closed) because client is allowed to negotiate // webrtc track before AddTrackRequest return to speed up the publishing process queued bool } +func (p *pendingTrackInfo) MarshalLogObject(e zapcore.ObjectEncoder) error { + if p == nil { + return nil + } + + e.AddArray("trackInfos", logger.ProtoSlice(p.trackInfos)) + e.AddArray("sdpRids", logger.StringSlice(p.sdpRids[:])) + e.AddBool("migrated", p.migrated) + e.AddTime("createdAt", p.createdAt) + e.AddBool("queued", p.queued) + return nil +} + +// -------------------------------------------------- + type pendingRemoteTrack struct { track *webrtc.TrackRemote receiver *webrtc.RTPReceiver @@ -920,28 +939,12 @@ func (p *ParticipantImpl) synthesizeAddTrackRequests(offer webrtc.SessionDescrip continue } - trackID := "" - - msid, ok := m.Attribute(sdp.AttrKeyMsid) - if ok { - if split := strings.Split(msid, " "); len(split) == 2 { - trackID = split[1] - } + cid := sdpHelper.GetMediaStreamTrack(m) + if cid == "" { + cid = guid.New(utils.TrackPrefix) } - if trackID == "" { - attr, ok := m.Attribute(sdp.AttrKeySSRC) - if ok { - split := strings.Split(attr, " ") - if len(split) == 3 && strings.HasPrefix(split[1], "msid:") { - trackID = split[2] - } - } - } - - if trackID == "" { - trackID = guid.New(utils.TrackPrefix) - } + rids, ridsOk := sdpHelper.GetSimulcastRids(m) var ( name string @@ -958,7 +961,7 @@ func (p *ParticipantImpl) synthesizeAddTrackRequests(offer webrtc.SessionDescrip trackType = livekit.TrackType_VIDEO } req := &livekit.AddTrackRequest{ - Cid: trackID, + Cid: cid, Name: name, Source: trackSource, Type: trackType, @@ -966,16 +969,67 @@ func (p *ParticipantImpl) synthesizeAddTrackRequests(offer webrtc.SessionDescrip Stereo: false, Stream: "camera", } - // ONE-SHOT-SIGNALLING-MODE-TODO: support video simulcast + // ONE-SHOT-SIGNALLING-MODE-TODO: simulcsat layer mapping if strings.EqualFold(m.MediaName.Media, "video") { - // dummy layer to ensure at least one layer is available - req.Layers = []*livekit.VideoLayer{{}} + if ridsOk { + // add simulcast layers, NOTE: only quality can be set as dimensions/fps is not available + n := min(len(rids), int(buffer.DefaultMaxLayerSpatial)+1) + for i := 0; i < n; i++ { + // WARN: casting int -> protobuf enum + req.Layers = append(req.Layers, &livekit.VideoLayer{Quality: livekit.VideoQuality(i)}) + } + } else { + // dummy layer to ensure at least one layer is available + req.Layers = []*livekit.VideoLayer{{}} + } } p.AddTrack(req) } return nil } +func (p *ParticipantImpl) updateRidsFromSDP(offer *webrtc.SessionDescription) { + parsed, err := offer.Unmarshal() + if err != nil { + return + } + + for _, m := range parsed.MediaDescriptions { + if m.MediaName.Media != "video" { + continue + } + + rids, ok := sdpHelper.GetSimulcastRids(m) + if !ok { + continue + } + + mst := sdpHelper.GetMediaStreamTrack(m) + if mst == "" { + continue + } + + p.pendingTracksLock.Lock() + pti := p.pendingTracks[mst] + if pti != nil { + // does not work for clients that use a different media stream track in SDP (e.g. Firefox) + // one option is to look up by track type, but that fails when there are multiple pending tracks + // of the same type + n := min(len(rids), len(pti.sdpRids)) + for i := 0; i < n; i++ { + pti.sdpRids[i] = rids[i] + } + + p.pubLogger.Debugw( + "pending track rids updated", + "trackID", pti.trackInfos[0].Sid, + "pendingTrack", pti, + ) + } + p.pendingTracksLock.Unlock() + } +} + // HandleOffer an offer from remote participant, used when clients make the initial connection func (p *ParticipantImpl) HandleOffer(offer webrtc.SessionDescription) error { p.pubLogger.Debugw("received offer", "transport", livekit.SignalTarget_PUBLISHER, "offer", offer) @@ -992,6 +1046,7 @@ func (p *ParticipantImpl) HandleOffer(offer webrtc.SessionDescription) error { } offer = p.setCodecPreferencesForPublisher(offer) + p.updateRidsFromSDP(&offer) err := p.TransportManager.HandleOffer(offer, shouldPend) if p.params.UseOneShotSignallingMode { if onSubscriberReady := p.getOnSubscriberReady(); onSubscriberReady != nil { @@ -1061,7 +1116,7 @@ func (p *ParticipantImpl) handleMigrateTracks() []*MediaTrack { p.pubLogger.Warnw("too many pending migrated tracks", nil, "trackID", pti.trackInfos[0].Sid, "count", len(pti.trackInfos), "cid", cid) } - mt := p.addMigratedTrack(cid, pti.trackInfos[0]) + mt := p.addMigratedTrack(cid, pti.trackInfos[0], pti.sdpRids) if mt != nil { addedTracks = append(addedTracks, mt) } else { @@ -1120,11 +1175,22 @@ func (p *ParticipantImpl) SetMigrateInfo( p.supervisor.SetPublicationMute(livekit.TrackID(ti.Sid), ti.Muted) } - p.pendingTracks[t.GetCid()] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, migrated: true, createdAt: time.Now()} - p.pubLogger.Infow("pending track added (migration)", "trackID", ti.Sid, "track", logger.Proto(ti)) + p.pendingTracks[t.GetCid()] = &pendingTrackInfo{ + trackInfos: []*livekit.TrackInfo{ti}, + sdpRids: buffer.DefaultVideoLayersRid, + migrated: true, + createdAt: time.Now(), + } + p.pubLogger.Infow( + "pending track added (migration)", + "trackID", ti.Sid, + "pendingTrack", p.pendingTracks[t.GetCid()], + ) } p.pendingTracksLock.Unlock() + p.updateRidsFromSDP(previousOffer) + if len(mediaTracks) != 0 { p.setIsPublisher(true) } @@ -2546,16 +2612,35 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l } if p.getPublishedTrackBySignalCid(req.Cid) != nil || p.getPublishedTrackBySdpCid(req.Cid) != nil || p.pendingTracks[req.Cid] != nil { if p.pendingTracks[req.Cid] == nil { - p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, createdAt: time.Now(), queued: true} + p.pendingTracks[req.Cid] = &pendingTrackInfo{ + trackInfos: []*livekit.TrackInfo{ti}, + sdpRids: buffer.DefaultVideoLayersRid, // could get updated from SDP + createdAt: time.Now(), + queued: true, + } } else { p.pendingTracks[req.Cid].trackInfos = append(p.pendingTracks[req.Cid].trackInfos, ti) } - p.pubLogger.Infow("pending track queued", "trackID", ti.Sid, "track", logger.Proto(ti), "request", logger.Proto(req)) + p.pubLogger.Infow( + "pending track queued", + "trackID", ti.Sid, + "request", logger.Proto(req), + "pendingTrack", p.pendingTracks[req.Cid], + ) return nil } - p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, createdAt: time.Now()} - p.pubLogger.Debugw("pending track added", "trackID", ti.Sid, "track", logger.Proto(ti), "request", logger.Proto(req)) + p.pendingTracks[req.Cid] = &pendingTrackInfo{ + trackInfos: []*livekit.TrackInfo{ti}, + sdpRids: buffer.DefaultVideoLayersRid, // could get updated from SDP + createdAt: time.Now(), + } + p.pubLogger.Debugw( + "pending track added", + "trackID", ti.Sid, + "request", logger.Proto(req), + "pendingTrack", p.pendingTracks[req.Cid], + ) return ti } @@ -2666,7 +2751,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track sfu.TrackRemote, rtpReceiver var isMigrated bool mt, ok := p.getPublishedTrackBySdpCid(track.ID()).(*MediaTrack) if !ok { - signalCid, ti, migrated, createdAt := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), true) + signalCid, ti, sdpRids, migrated, createdAt := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind()), true) if ti == nil { p.pendingRemoteTracks = append( p.pendingRemoteTracks, @@ -2706,7 +2791,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track sfu.TrackRemote, rtpReceiver // only assign version on a fresh publish, i. e. avoid updating version in scenarios like migration ti.Version = p.params.VersionGenerator.Next().ToProto() } - mt = p.addMediaTrack(signalCid, track.ID(), ti) + mt = p.addMediaTrack(signalCid, track.ID(), ti, sdpRids) newTrack = true // if the addTrackRequest is sent before participant active then it means the client tries to publish @@ -2754,7 +2839,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track sfu.TrackRemote, rtpReceiver return mt, newTrack } -func (p *ParticipantImpl) addMigratedTrack(cid string, ti *livekit.TrackInfo) *MediaTrack { +func (p *ParticipantImpl) addMigratedTrack(cid string, ti *livekit.TrackInfo, sdpRids buffer.VideoLayersRid) *MediaTrack { p.pubLogger.Infow("add migrated track", "cid", cid, "trackID", ti.Sid, "track", logger.Proto(ti)) rtpReceiver := p.TransportManager.GetPublisherRTPReceiver(ti.Mid) if rtpReceiver == nil { @@ -2762,7 +2847,7 @@ func (p *ParticipantImpl) addMigratedTrack(cid string, ti *livekit.TrackInfo) *M return nil } - mt := p.addMediaTrack(cid, cid, ti) + mt := p.addMediaTrack(cid, cid, ti, sdpRids) potentialCodecs := make([]webrtc.RTPCodecParameters, 0, len(ti.Codecs)) parameters := rtpReceiver.GetParameters() @@ -2807,7 +2892,7 @@ func (p *ParticipantImpl) addMigratedTrack(cid string, ti *livekit.TrackInfo) *M return mt } -func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *livekit.TrackInfo) *MediaTrack { +func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *livekit.TrackInfo, sdpRids buffer.VideoLayersRid) *MediaTrack { mt := NewMediaTrack(MediaTrackParams{ SignalCid: signalCid, SdpCid: sdpCid, @@ -2830,6 +2915,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, sdpCid string, ti *liv ShouldRegressCodec: func() bool { return p.helper().ShouldRegressCodec() }, + Rids: sdpRids, }, ti) mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) @@ -2939,7 +3025,7 @@ func (p *ParticipantImpl) onUpTrackManagerClose() { p.pubRTCPQueue.Stop() } -func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType, skipQueued bool) (string, *livekit.TrackInfo, bool, time.Time) { +func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType, skipQueued bool) (string, *livekit.TrackInfo, buffer.VideoLayersRid, bool, time.Time) { signalCid := clientId pendingInfo := p.pendingTracks[clientId] if pendingInfo == nil { @@ -2974,10 +3060,10 @@ func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackTyp // if still not found, we are done if pendingInfo == nil || (skipQueued && pendingInfo.queued) { - return signalCid, nil, false, time.Time{} + return signalCid, nil, buffer.VideoLayersRid{}, false, time.Time{} } - return signalCid, utils.CloneProto(pendingInfo.trackInfos[0]), pendingInfo.migrated, pendingInfo.createdAt + return signalCid, utils.CloneProto(pendingInfo.trackInfos[0]), pendingInfo.sdpRids, pendingInfo.migrated, pendingInfo.createdAt } // setTrackID either generates a new TrackID for an AddTrackRequest diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index b6d1a1e1d..71ecf266f 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -289,7 +289,7 @@ func TestMuteSetting(t *testing.T) { Muted: true, }) - _, ti, _, _ := p.getPendingTrack("cid", livekit.TrackType_AUDIO, false) + _, ti, _, _, _ := p.getPendingTrack("cid", livekit.TrackType_AUDIO, false) require.NotNil(t, ti) require.True(t, ti.Muted) }) diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index 401c1add6..046e6f710 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -46,7 +46,7 @@ func (p *ParticipantImpl) setCodecPreferencesOpusRedForPublisher(offer webrtc.Se } p.pendingTracksLock.RLock() - _, info, _, _ := p.getPendingTrack(streamID, livekit.TrackType_AUDIO, false) + _, info, _, _, _ := p.getPendingTrack(streamID, livekit.TrackType_AUDIO, false) // if RED is disabled for this track, don't prefer RED codec in offer disableRed := info != nil && info.DisableRed p.pendingTracksLock.RUnlock() @@ -132,7 +132,7 @@ func (p *ParticipantImpl) setCodecPreferencesVideoForPublisher(offer webrtc.Sess if mt != nil { info = mt.ToProto() } else { - _, info, _, _ = p.getPendingTrack(streamID, livekit.TrackType_VIDEO, false) + _, info, _, _, _ = p.getPendingTrack(streamID, livekit.TrackType_VIDEO, false) } if info == nil { @@ -227,7 +227,7 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript track, _ := p.getPublishedTrackBySdpCid(streamID).(*MediaTrack) if track == nil { p.pendingTracksLock.RLock() - _, ti, _, _ = p.getPendingTrack(streamID, livekit.TrackType_AUDIO, false) + _, ti, _, _, _ = p.getPendingTrack(streamID, livekit.TrackType_AUDIO, false) p.pendingTracksLock.RUnlock() } else { ti = track.ToProto() diff --git a/pkg/sfu/buffer/videolayerutils.go b/pkg/sfu/buffer/videolayerutils.go index f099c2e7c..cdf6c986e 100644 --- a/pkg/sfu/buffer/videolayerutils.go +++ b/pkg/sfu/buffer/videolayerutils.go @@ -25,6 +25,12 @@ const ( FullResolution = "f" ) +type VideoLayersRid [DefaultMaxLayerSpatial + 1]string + +var ( + DefaultVideoLayersRid = VideoLayersRid{QuarterResolution, HalfResolution, FullResolution} +) + // SIMULCAST-CODEC-TODO: these need to be codec mime aware if and when each codec suppports different layers func LayerPresenceFromTrackInfo(trackInfo *livekit.TrackInfo) *[livekit.VideoQuality_HIGH + 1]bool { if trackInfo == nil || len(trackInfo.Layers) == 0 { @@ -44,7 +50,7 @@ func LayerPresenceFromTrackInfo(trackInfo *livekit.TrackInfo) *[livekit.VideoQua return &layerPresence } -func RidToSpatialLayer(rid string, trackInfo *livekit.TrackInfo) int32 { +func RidToSpatialLayer(rid string, trackInfo *livekit.TrackInfo, ridSpace VideoLayersRid) int32 { lp := LayerPresenceFromTrackInfo(trackInfo) if lp == nil { switch rid { @@ -60,7 +66,7 @@ func RidToSpatialLayer(rid string, trackInfo *livekit.TrackInfo) int32 { } switch rid { - case QuarterResolution: + case ridSpace[0]: switch { case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: fallthrough @@ -76,7 +82,7 @@ func RidToSpatialLayer(rid string, trackInfo *livekit.TrackInfo) int32 { return 0 } - case HalfResolution: + case ridSpace[1]: switch { case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: fallthrough @@ -92,7 +98,7 @@ func RidToSpatialLayer(rid string, trackInfo *livekit.TrackInfo) int32 { return 0 } - case FullResolution: + case ridSpace[2]: switch { case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: return 2 @@ -118,7 +124,7 @@ func RidToSpatialLayer(rid string, trackInfo *livekit.TrackInfo) int32 { } } -func SpatialLayerToRid(layer int32, trackInfo *livekit.TrackInfo) string { +func SpatialLayerToRid(layer int32, trackInfo *livekit.TrackInfo, ridSpace VideoLayersRid) string { lp := LayerPresenceFromTrackInfo(trackInfo) if lp == nil { switch layer { @@ -143,10 +149,10 @@ func SpatialLayerToRid(layer int32, trackInfo *livekit.TrackInfo) string { case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_HIGH]: fallthrough case lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: - return QuarterResolution + return ridSpace[0] default: - return QuarterResolution + return ridSpace[0] } case 1: @@ -158,38 +164,38 @@ func SpatialLayerToRid(layer int32, trackInfo *livekit.TrackInfo) string { case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_HIGH]: fallthrough case lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: - return HalfResolution + return ridSpace[1] default: - return QuarterResolution + return ridSpace[0] } case 2: switch { case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: - return FullResolution + return ridSpace[2] case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_MEDIUM]: logger.Warnw("unexpected layer 2 with only two qualities, low and medium", nil, "trackID", trackInfo.Sid, "trackInfo", logger.Proto(trackInfo)) - return HalfResolution + return ridSpace[1] case lp[livekit.VideoQuality_LOW] && lp[livekit.VideoQuality_HIGH]: logger.Warnw("unexpected layer 2 with only two qualities, low and high", nil, "trackID", trackInfo.Sid, "trackInfo", logger.Proto(trackInfo)) - return HalfResolution + return ridSpace[1] case lp[livekit.VideoQuality_MEDIUM] && lp[livekit.VideoQuality_HIGH]: logger.Warnw("unexpected layer 2 with only two qualities, medium and high", nil, "trackID", trackInfo.Sid, "trackInfo", logger.Proto(trackInfo)) - return HalfResolution + return ridSpace[1] default: - return QuarterResolution + return ridSpace[0] } default: - return QuarterResolution + return ridSpace[0] } } -func VideoQualityToRid(quality livekit.VideoQuality, trackInfo *livekit.TrackInfo) string { - return SpatialLayerToRid(VideoQualityToSpatialLayer(quality, trackInfo), trackInfo) +func VideoQualityToRid(quality livekit.VideoQuality, trackInfo *livekit.TrackInfo, ridSpace VideoLayersRid) string { + return SpatialLayerToRid(VideoQualityToSpatialLayer(quality, trackInfo), trackInfo, ridSpace) } func SpatialLayerToVideoQuality(layer int32, trackInfo *livekit.TrackInfo) livekit.VideoQuality { diff --git a/pkg/sfu/buffer/videolayerutils_test.go b/pkg/sfu/buffer/videolayerutils_test.go index bb103f72c..365b83a5b 100644 --- a/pkg/sfu/buffer/videolayerutils_test.go +++ b/pkg/sfu/buffer/videolayerutils_test.go @@ -160,10 +160,10 @@ func TestRidConversion(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { for testRid, expectedResult := range test.ridToLayer { - actualLayer := RidToSpatialLayer(testRid, test.trackInfo) + actualLayer := RidToSpatialLayer(testRid, test.trackInfo, DefaultVideoLayersRid) require.Equal(t, expectedResult.layer, actualLayer) - actualRid := SpatialLayerToRid(actualLayer, test.trackInfo) + actualRid := SpatialLayerToRid(actualLayer, test.trackInfo, DefaultVideoLayersRid) require.Equal(t, expectedResult.rid, actualRid) } }) @@ -434,7 +434,7 @@ func TestVideoQualityToRidConversion(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { for testQuality, expectedRid := range test.qualityToRid { - actualRid := VideoQualityToRid(testQuality, test.trackInfo) + actualRid := VideoQualityToRid(testQuality, test.trackInfo, DefaultVideoLayersRid) require.Equal(t, expectedRid, actualRid) } }) diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 3f4fd9eff..6a2f69cbd 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -177,6 +177,7 @@ type WebRTCReceiver struct { closed atomic.Bool useTrackers bool trackInfo atomic.Pointer[livekit.TrackInfo] + rids buffer.VideoLayersRid onRTCP func([]rtcp.Packet) @@ -251,6 +252,7 @@ func NewWebRTCReceiver( receiver *webrtc.RTPReceiver, track TrackRemote, trackInfo *livekit.TrackInfo, + rids buffer.VideoLayersRid, logger logger.Logger, onRTCP func([]rtcp.Packet), streamTrackerManagerConfig StreamTrackerManagerConfig, @@ -264,6 +266,7 @@ func NewWebRTCReceiver( codec: track.Codec(), codecState: ReceiverCodecStateNormal, kind: track.Kind(), + rids: rids, onRTCP: onRTCP, isSVC: mime.IsMimeTypeStringSVC(track.Codec().MimeType), isRED: mime.IsMimeTypeStringRED(track.Codec().MimeType), @@ -401,7 +404,7 @@ func (w *WebRTCReceiver) AddUpTrack(track TrackRemote, buff *buffer.Buffer) erro layer := int32(0) if w.Kind() == webrtc.RTPCodecTypeVideo && !w.isSVC { - layer = buffer.RidToSpatialLayer(track.RID(), w.trackInfo.Load()) + layer = buffer.RidToSpatialLayer(track.RID(), w.trackInfo.Load(), w.rids) } buff.SetLogger(w.logger.WithValues("layer", layer)) buff.SetAudioLevelParams(audio.AudioLevelParams{