use synchronized GetParticipants to ensure eliminate timing issues upon participant join

cmd/cli/client/client.go

@@ -197,7 +197,7 @@ func (c *RTCClient) Run() error {
 			}
 			c.lock.Unlock()
 
-			logger.Debugw("join accepted, sending offer..", "participant", msg.Join.Participant.Sid)
+			logger.Debugw("join accepted, sending offer..", "participant", msg.Join.Participant.Identity)
 			c.localParticipant = msg.Join.Participant
 			logger.Debugw("other participants", "count", len(msg.Join.OtherParticipants))
 
@@ -247,7 +247,7 @@ func (c *RTCClient) Run() error {
 			c.lock.Lock()
 			for _, p := range msg.Update.Participants {
 				c.remoteParticipants[p.Sid] = p
-				logger.Debugw("participant update", "id", p.Sid, "state", p.State.String())
+				logger.Debugw("participant update", "id", p.Identity, "state", p.State.String())
 			}
 			c.lock.Unlock()
 		case *livekit.SignalResponse_TrackPublished:

pkg/rtc/mediatrack.go

@@ -153,7 +153,7 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error {
 			logger.Debugw("removing peerconnection track",
 				"track", t.id,
 				"srcParticipant", t.participantId,
-				"destParticipant", participant.ID())
+				"destParticipant", participant.Identity())
 			if err := participant.PeerConnection().RemoveTrack(sender); err != nil {
 				if err == webrtc.ErrConnectionClosed {
 					// participant closing, can skip removing downtracks

pkg/rtc/participant.go

@@ -354,8 +354,8 @@ func (p *ParticipantImpl) AddSubscriber(op types.Participant) error {
 
 	for _, track := range p.publishedTracks {
 		logger.Debugw("subscribing to remoteTrack",
-			"srcParticipant", p.ID(),
-			"dstParticipant", op.ID(),
+			"srcParticipant", p.Identity(),
+			"dstParticipant", op.Identity(),
 			"remoteTrack", track.ID())
 		if err := track.AddSubscriber(op); err != nil {
 			return err
@@ -496,7 +496,7 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
 		return
 	}
 	p.state.Store(state)
-	logger.Debugw("updating participant state", "state", state.String(), "participant", p.ID())
+	logger.Debugw("updating participant state", "state", state.String(), "participant", p.Identity())
 	if p.onStateChange != nil {
 		go func() {
 			defer Recover()
@@ -507,7 +507,7 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) {
 
 // when a new remoteTrack is created, creates a Track and adds it to room
 func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) {
-	logger.Debugw("mediaTrack added", "participantId", p.ID(), "remoteTrack", track.ID())
+	logger.Debugw("mediaTrack added", "participant", p.Identity(), "remoteTrack", track.ID())
 
 	ti := p.popPendingTrack(track.ID())
 	if ti == nil {
@@ -526,7 +526,7 @@ func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) {
 	if dc.Label() == placeholderDataChannel {
 		return
 	}
-	logger.Debugw("dataChannel added", "participantId", p.ID(), "label", dc.Label())
+	logger.Debugw("dataChannel added", "participant", p.Identity(), "label", dc.Label())
 
 	// data channels have numeric ids, so we use its label to identify
 	ti := p.popPendingTrack(dc.Label())
@@ -616,7 +616,7 @@ func (p *ParticipantImpl) downTracksRTCPWorker() {
 					return
 				}
 				logger.Errorw("could not send downtrack reports",
-					"participant", p.id,
+					"participant", p.Identity(),
 					"err", err)
 			}
 			pkts = pkts[:0]
@@ -640,7 +640,7 @@ func (p *ParticipantImpl) rtcpSendWorker() {
 		//}
 		if err := p.peerConn.WriteRTCP(pkts); err != nil {
 			logger.Errorw("could not write RTCP to participant",
-				"participant", p.id,
+				"participant", p.Identity(),
 				"err", err)
 		}
 	}

pkg/rtc/room.go

@@ -98,13 +98,13 @@ func (r *Room) Join(participant types.Participant) error {
 	// it's important to set this before connection, we don't want to miss out on any publishedTracks
 	participant.OnTrackPublished(r.onTrackAdded)
 	participant.OnStateChange(func(p types.Participant, oldState livekit.ParticipantInfo_State) {
-		logger.Debugw("participant state changed", "state", p.State(), "participant", p.ID(),
+		logger.Debugw("participant state changed", "state", p.State(), "participant", p.Identity(),
 			"oldState", oldState)
 		r.broadcastParticipantState(p)
 
 		if oldState == livekit.ParticipantInfo_JOINING && p.State() == livekit.ParticipantInfo_JOINED {
 			// subscribe participant to existing publishedTracks
-			for _, op := range r.participants {
+			for _, op := range r.GetParticipants() {
 				if p.ID() == op.ID() {
 					// don't send to itself
 					continue

pkg/service/rtcservice.go

@@ -1,7 +1,6 @@
 package service
 
 import (
-	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -153,14 +152,6 @@ type errStruct struct {
 	Message    string `json:"message,omitempty"`
 }
 
-func writeJSONError(w http.ResponseWriter, code int, error ...string) {
-	w.Header().Set("Content-Type", "application/json; charset=utf-8")
-	w.Header().Set("X-Content-Type-Options", "nosniff")
-	w.WriteHeader(code)
-
-	json.NewEncoder(w).Encode(jsonError(code, error...))
-}
-
 func jsonError(code int, error ...string) errStruct {
 	es := errStruct{
 		StatusCode: code,