Support dynacast. (#690)

* Support dynacast.

* get mime from simulcast codec

* clean up

* comments

* logging

* use next sample time as it is adjusted for sample duration

Author: boks1971

engine.go

@@ -69,30 +69,31 @@ type RTCEngine struct {
 	JoinTimeout time.Duration
 
 	// callbacks
-	OnLocalTrackUnpublished func(response *livekit.TrackUnpublishedResponse)
-	OnTrackRemoteMuted      func(request *livekit.MuteTrackRequest)
-	OnDisconnected          func(reason DisconnectionReason)
-	OnMediaTrack            func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver)
-	OnParticipantUpdate     func([]*livekit.ParticipantInfo)
-	OnSpeakersChanged       func([]*livekit.SpeakerInfo)
-	OnDataReceived          func(userPacket *livekit.UserPacket) // Deprecated: Use OnDataPacket instead
-	OnDataPacket            func(identity string, dataPacket DataPacket)
-	OnConnectionQuality     func([]*livekit.ConnectionQualityInfo)
-	OnRoomUpdate            func(room *livekit.Room)
-	OnRoomMoved             func(moved *livekit.RoomMovedResponse)
-	OnRestarting            func()
-	OnRestarted             func(*livekit.JoinResponse)
-	OnResuming              func()
-	OnResumed               func()
-	OnTranscription         func(*livekit.Transcription)
-	OnSignalClientConnected func(*livekit.JoinResponse)
-	OnRpcRequest            func(callerIdentity, requestId, method, payload string, responseTimeout time.Duration, version uint32)
-	OnRpcAck                func(requestId string)
-	OnRpcResponse           func(requestId string, payload *string, error *RpcError)
-	OnStreamHeader          func(*livekit.DataStream_Header, string)
-	OnStreamChunk           func(*livekit.DataStream_Chunk)
-	OnStreamTrailer         func(*livekit.DataStream_Trailer)
-	OnLocalTrackSubscribed  func(trackSubscribed *livekit.TrackSubscribed)
+	OnLocalTrackUnpublished   func(response *livekit.TrackUnpublishedResponse)
+	OnTrackRemoteMuted        func(request *livekit.MuteTrackRequest)
+	OnDisconnected            func(reason DisconnectionReason)
+	OnMediaTrack              func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver)
+	OnParticipantUpdate       func([]*livekit.ParticipantInfo)
+	OnSpeakersChanged         func([]*livekit.SpeakerInfo)
+	OnDataReceived            func(userPacket *livekit.UserPacket) // Deprecated: Use OnDataPacket instead
+	OnDataPacket              func(identity string, dataPacket DataPacket)
+	OnConnectionQuality       func([]*livekit.ConnectionQualityInfo)
+	OnRoomUpdate              func(room *livekit.Room)
+	OnRoomMoved               func(moved *livekit.RoomMovedResponse)
+	OnRestarting              func()
+	OnRestarted               func(*livekit.JoinResponse)
+	OnResuming                func()
+	OnResumed                 func()
+	OnTranscription           func(*livekit.Transcription)
+	OnSignalClientConnected   func(*livekit.JoinResponse)
+	OnRpcRequest              func(callerIdentity, requestId, method, payload string, responseTimeout time.Duration, version uint32)
+	OnRpcAck                  func(requestId string)
+	OnRpcResponse             func(requestId string, payload *string, error *RpcError)
+	OnStreamHeader            func(*livekit.DataStream_Header, string)
+	OnStreamChunk             func(*livekit.DataStream_Chunk)
+	OnStreamTrailer           func(*livekit.DataStream_Trailer)
+	OnLocalTrackSubscribed    func(trackSubscribed *livekit.TrackSubscribed)
+	OnSubscribedQualityUpdate func(subscribedQualityUpdate *livekit.SubscribedQualityUpdate)
 
 	onClose     []func()
 	onCloseLock sync.Mutex
@@ -145,6 +146,11 @@ func NewRTCEngine() *RTCEngine {
 			f(trackSubscribed)
 		}
 	}
+	e.client.OnSubscribedQualityUpdate = func(subscribedQualityUpdate *livekit.SubscribedQualityUpdate) {
+		if f := e.OnSubscribedQualityUpdate; f != nil {
+			f(subscribedQualityUpdate)
+		}
+	}
 	e.client.OnClose = func() { e.handleDisconnect(false) }
 	e.onClose = []func(){}
 

localtrack.go

@@ -359,7 +359,7 @@ func (s *LocalTrack) WriteSample(sample media.Sample, opts *SampleWriteOptions)
 	//   4. PrevDroppedPackets -> number of dropped packets before this sample
 	//
 	// The goal here is to calculate RTP time stamp of provided sample.
-	// Priority of what is used (eevn if multiple are provided)
+	// Priority of what is used (even if multiple are provided)
 	//   1. PacketTimestamp
 	//   2. Timestamp
 	//   3. Duration
@@ -602,6 +602,7 @@ func (s *LocalTrack) writeWorker(provider SampleProvider, onComplete func()) {
 				}
 			}
 
+			sample.Timestamp = nextSampleTime
 			if err := s.WriteSample(sample, opts); err != nil {
 				s.log.Errorw("could not write sample", err)
 				return

publication.go

@@ -72,8 +72,31 @@ func (p *trackPublicationBase) Track() Track {
 }
 
 func (p *trackPublicationBase) MimeType() string {
+	// This requires some more work.
+	// TrackInfo has a top level MimeType which is not set
+	// on server side till the track is published.
+	// So, it is not available in the TrackPublishedResponse.
+	//
+	// But, if client specified SimulcastCodecs in AddTrackRequest,
+	// the TrackPublishedResponse will have Codecs populated and
+	// that will have the MimeType specified in AddTrackRequest.
+	// Just taking the first one here which has a non-nil MimeType.
+	// This is okay (for tracks published from here)
+	// as of 2025-05-12, 1:30 pm Pacific as
+	// Go SDK does not (yet) support simulcast codec feature.
+	//
+	// When simulcast codec feature is added, this struct needs
+	// to be updated to handle multiple mime types
 	if info, ok := p.info.Load().(*livekit.TrackInfo); ok {
-		return info.MimeType
+		if info.MimeType != "" {
+			return info.MimeType
+		}
+
+		for _, codec := range info.Codecs {
+			if codec.MimeType != "" {
+				return codec.MimeType
+			}
+		}
 	}
 	return ""
 }

remoteparticipant.go

@@ -93,8 +93,11 @@ func (p *RemoteParticipant) updateInfo(pi *livekit.ParticipantInfo) {
 	}
 }
 
-func (p *RemoteParticipant) addSubscribedMediaTrack(track *webrtc.TrackRemote, trackSID string,
-	receiver *webrtc.RTPReceiver) {
+func (p *RemoteParticipant) addSubscribedMediaTrack(
+	track *webrtc.TrackRemote,
+	trackSID string,
+	receiver *webrtc.RTPReceiver,
+) {
 	pub := p.getPublication(trackSID)
 	if pub == nil {
 		// wait for metadata to arrive

room.go

@@ -238,6 +238,7 @@ func NewRoom(callback *RoomCallback) *Room {
 	engine.OnStreamChunk = r.handleStreamChunk
 	engine.OnStreamTrailer = r.handleStreamTrailer
 	engine.OnLocalTrackSubscribed = r.handleLocalTrackSubscribed
+	engine.OnSubscribedQualityUpdate = r.handleSubscribedQualityUpdate
 
 	// callbacks engine can use to get data
 	engine.CbGetLocalParticipantSID = r.getLocalParticipantSID
@@ -843,6 +844,39 @@ func (r *Room) handleLocalTrackSubscribed(trackSubscribed *livekit.TrackSubscrib
 	r.callback.OnLocalTrackSubscribed(trackPublication, r.LocalParticipant)
 }
 
+func (r *Room) handleSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate) {
+	trackPublication := r.LocalParticipant.getLocalPublication(subscribedQualityUpdate.TrackSid)
+	if trackPublication == nil {
+		r.log.Debugw("recieved subscribed quality update for unknown track", "trackID", subscribedQualityUpdate.TrackSid)
+		return
+	}
+
+	r.log.Infow(
+		"handling subscribed quality update",
+		"trackID", trackPublication.SID(),
+		"mime", trackPublication.MimeType(),
+		"subscribedQualityUpdate", protoLogger.Proto(subscribedQualityUpdate),
+	)
+	for _, subscribedCodec := range subscribedQualityUpdate.SubscribedCodecs {
+		if !strings.HasSuffix(strings.ToLower(trackPublication.MimeType()), subscribedCodec.Codec) {
+			continue
+		}
+
+		for _, subscribedQuality := range subscribedCodec.Qualities {
+			track := trackPublication.GetSimulcastTrack(subscribedQuality.Quality)
+			if track != nil {
+				track.setMuted(!subscribedQuality.Enabled)
+				r.log.Infow(
+					"updating layer enable",
+					"trackID", trackPublication.SID(),
+					"quality", subscribedQuality.Quality,
+					"enabled", subscribedQuality.Enabled,
+				)
+			}
+		}
+	}
+}
+
 func (r *Room) sendSyncState() {
 	subscriber, ok := r.engine.Subscriber()
 	if !ok || subscriber.pc.RemoteDescription() == nil {

signalclient.go

@@ -48,21 +48,22 @@ type SignalClient struct {
 	pendingResponse *livekit.SignalResponse
 	readerClosedCh  chan struct{}
 
-	OnClose                 func()
-	OnAnswer                func(sd webrtc.SessionDescription)
-	OnOffer                 func(sd webrtc.SessionDescription)
-	OnTrickle               func(init webrtc.ICECandidateInit, target livekit.SignalTarget)
-	OnParticipantUpdate     func([]*livekit.ParticipantInfo)
-	OnLocalTrackPublished   func(response *livekit.TrackPublishedResponse)
-	OnSpeakersChanged       func([]*livekit.SpeakerInfo)
-	OnConnectionQuality     func([]*livekit.ConnectionQualityInfo)
-	OnRoomUpdate            func(room *livekit.Room)
-	OnRoomMoved             func(moved *livekit.RoomMovedResponse)
-	OnTrackRemoteMuted      func(request *livekit.MuteTrackRequest)
-	OnLocalTrackUnpublished func(response *livekit.TrackUnpublishedResponse)
-	OnTokenRefresh          func(refreshToken string)
-	OnLeave                 func(*livekit.LeaveRequest)
-	OnLocalTrackSubscribed  func(trackSubscribed *livekit.TrackSubscribed)
+	OnClose                   func()
+	OnAnswer                  func(sd webrtc.SessionDescription)
+	OnOffer                   func(sd webrtc.SessionDescription)
+	OnTrickle                 func(init webrtc.ICECandidateInit, target livekit.SignalTarget)
+	OnParticipantUpdate       func([]*livekit.ParticipantInfo)
+	OnLocalTrackPublished     func(response *livekit.TrackPublishedResponse)
+	OnSpeakersChanged         func([]*livekit.SpeakerInfo)
+	OnConnectionQuality       func([]*livekit.ConnectionQualityInfo)
+	OnRoomUpdate              func(room *livekit.Room)
+	OnRoomMoved               func(moved *livekit.RoomMovedResponse)
+	OnTrackRemoteMuted        func(request *livekit.MuteTrackRequest)
+	OnLocalTrackUnpublished   func(response *livekit.TrackUnpublishedResponse)
+	OnTokenRefresh            func(refreshToken string)
+	OnLeave                   func(*livekit.LeaveRequest)
+	OnLocalTrackSubscribed    func(trackSubscribed *livekit.TrackSubscribed)
+	OnSubscribedQualityUpdate func(subscribedQualityUpdate *livekit.SubscribedQualityUpdate)
 }
 
 func NewSignalClient() *SignalClient {
@@ -422,6 +423,10 @@ func (c *SignalClient) handleResponse(res *livekit.SignalResponse) {
 		if c.OnLocalTrackSubscribed != nil {
 			c.OnLocalTrackSubscribed(msg.TrackSubscribed)
 		}
+	case *livekit.SignalResponse_SubscribedQualityUpdate:
+		if c.OnSubscribedQualityUpdate != nil {
+			c.OnSubscribedQualityUpdate(msg.SubscribedQualityUpdate)
+		}
 	}
 }