Minor changes to publisher bool. (#1880)

* Minor changes to publisher bool.

* address feedback
This commit is contained in:
Raja Subramanian
2023-07-15 12:43:05 +05:30
committed by GitHub
parent 4c02a6d717
commit 469f1cd073
2 changed files with 33 additions and 31 deletions
+32 -29
View File
@@ -370,13 +370,14 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio
return false
}
p.GetLogger().Infow("updating participant permission", "permission", permission)
p.params.Logger.Infow("updating participant permission", "permission", permission)
video.UpdateFromPermission(permission)
p.dirty.Store(true)
canPublish := video.GetCanPublish()
canSubscribe := video.GetCanSubscribe()
onParticipantUpdate := p.onParticipantUpdate
onClaimsChanged := p.onClaimsChanged
@@ -387,13 +388,7 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio
// publish permission has been revoked then remove offending tracks
for _, track := range p.GetPublishedTracks() {
if !video.GetCanPublishSource(track.Source()) {
p.RemovePublishedTrack(track, false, false)
if p.ProtocolVersion().SupportsUnpublish() {
p.sendTrackUnpublished(track.ID())
} else {
// for older clients that don't support unpublish, mute to avoid them sending data
p.sendTrackMuted(track.ID(), true)
}
p.removePublishedTrack(track)
}
}
@@ -1190,22 +1185,24 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
}
func (p *ParticipantImpl) setIsPublisher(isPublisher bool) {
if p.isPublisher.Swap(isPublisher) != isPublisher {
p.lock.Lock()
p.requireBroadcast = true
p.lock.Unlock()
if p.isPublisher.Swap(isPublisher) == isPublisher {
return
}
p.dirty.Store(true)
p.lock.Lock()
p.requireBroadcast = true
p.lock.Unlock()
// trigger update as well if participant is already fully connected
if p.State() == livekit.ParticipantInfo_ACTIVE {
p.lock.RLock()
onParticipantUpdate := p.onParticipantUpdate
p.lock.RUnlock()
p.dirty.Store(true)
if onParticipantUpdate != nil {
onParticipantUpdate(p)
}
// trigger update as well if participant is already fully connected
if p.State() == livekit.ParticipantInfo_ACTIVE {
p.lock.RLock()
onParticipantUpdate := p.onParticipantUpdate
p.lock.RUnlock()
if onParticipantUpdate != nil {
onParticipantUpdate(p)
}
}
}
@@ -1220,6 +1217,16 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription) err
})
}
func (p *ParticipantImpl) removePublishedTrack(track types.MediaTrack) {
p.RemovePublishedTrack(track, false, false)
if p.ProtocolVersion().SupportsUnpublish() {
p.sendTrackUnpublished(track.ID())
} else {
// for older clients that don't support unpublish, mute to avoid them sending data
p.sendTrackMuted(track.ID(), true)
}
}
// when a new remoteTrack is created, creates a Track and adds it to room
func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
if p.IsDisconnected() {
@@ -1242,12 +1249,12 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
p.params.Logger.Warnw("no permission to publish mediaTrack", nil,
"source", publishedTrack.Source(),
)
p.removePublishedTrack(publishedTrack)
return
}
if !p.IsPublisher() {
p.setIsPublisher(true)
}
p.setIsPublisher(true)
p.dirty.Store(true)
p.params.Logger.Infow("mediaTrack published",
"kind", track.Kind().String(),
@@ -1258,8 +1265,6 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w
"mime", track.Codec().MimeType,
)
p.dirty.Store(true)
if !isNewTrack && !publishedTrack.HasPendingCodec() && p.IsReady() {
p.lock.RLock()
onTrackUpdated := p.onTrackUpdated
@@ -1300,9 +1305,7 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt
p.params.Logger.Warnw("received unsupported data packet", nil, "payload", payload)
}
if !p.IsPublisher() {
p.setIsPublisher(true)
}
p.setIsPublisher(true)
}
func (p *ParticipantImpl) onICECandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) error {
+1 -2
View File
@@ -828,7 +828,6 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS
func (r *Room) onTrackPublished(participant types.LocalParticipant, track types.MediaTrack) {
// publish participant update, since track state is changed
r.broadcastParticipantState(participant, broadcastOptions{skipSource: true})
r.protoProxy.MarkDirty(false)
r.lock.RLock()
// subscribe all existing participants to this MediaTrack
@@ -888,7 +887,6 @@ func (r *Room) onTrackUpdated(p types.LocalParticipant, _ types.MediaTrack) {
func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTrack) {
r.trackManager.RemoveTrack(track)
r.protoProxy.MarkDirty(false)
if !p.IsClosed() {
r.broadcastParticipantState(p, broadcastOptions{skipSource: true})
}
@@ -898,6 +896,7 @@ func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTra
}
func (r *Room) onParticipantUpdate(p types.LocalParticipant) {
r.protoProxy.MarkDirty(false)
// immediately notify when permissions or metadata changed
r.broadcastParticipantState(p, broadcastOptions{immediate: true})
if r.onParticipantChanged != nil {