From 469f1cd073b2806c463a94f3831bba5ae4aa6748 Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Sat, 15 Jul 2023 12:43:05 +0530
Subject: [PATCH] Minor changes to publisher bool. (#1880)

* Minor changes to publisher bool.

* address feedback
---
 pkg/rtc/participant.go | 61 ++++++++++++++++++++++--------------------
 pkg/rtc/room.go        |  3 +--
 2 files changed, 33 insertions(+), 31 deletions(-)

diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go
index 5de6904c1..e0d5214d4 100644
--- a/pkg/rtc/participant.go
+++ b/pkg/rtc/participant.go
@@ -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 {
diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go
index a90b90676..7602659ac 100644
--- a/pkg/rtc/room.go
+++ b/pkg/rtc/room.go
@@ -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 {