Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Reconcile track mute state by client to avoid mute message loop #352

Merged
merged 1 commit into from
Nov 20, 2023
Merged

Reconcile track mute state by client to avoid mute message loop #352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 13 additions & 10 deletions localparticipant.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ func (p *LocalParticipant) PublishTrack(track webrtc.TrackLocal, opts *TrackPubl
pub.OnRttUpdate(func(rtt uint32) {
p.engine.setRTT(rtt)
})
pub.onMuteChanged = p.onTrackMuted

req := &livekit.AddTrackRequest{
Cid: track.ID(),
Expand Down Expand Up @@ -149,6 +150,7 @@ func (p *LocalParticipant) PublishSimulcastTrack(tracks []*LocalSampleTrack, opt
mainTrack := tracks[len(tracks)-1]

pub := NewLocalTrackPublication(KindFromRTPType(mainTrack.Kind()), nil, *opts, p.engine.client)
pub.onMuteChanged = p.onTrackMuted

var layers []*livekit.VideoLayer
for _, st := range tracks {
Expand Down Expand Up @@ -359,16 +361,7 @@ func (p *LocalParticipant) updateInfo(info *livekit.ParticipantInfo) {
continue
}
if pub.IsMuted() != ti.Muted {
pub.SetMuted(ti.Muted)

// trigger callback
if ti.Muted {
p.Callback.OnTrackMuted(pub, p)
p.roomCallback.OnTrackMuted(pub, p)
} else if !ti.Muted {
p.Callback.OnTrackUnmuted(pub, p)
p.roomCallback.OnTrackUnmuted(pub, p)
}
_ = p.engine.client.SendMuteTrack(pub.SID(), pub.IsMuted())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be called directly? Feels like pub.SetMuted would be the right thing as it will do the right thing of sending message and issuing callback? Did you want to remove the callback when mute is managed through this path? Even then, calling SetMuted and doing callback or not there might be better?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The pub.SetMuted will find the new state is the same as the current publication state and return directly, this pr keeps same behavior like js-client to send a mute message only when server-side info is inconsistent

}
}
}
Expand All @@ -379,3 +372,13 @@ func (p *LocalParticipant) getLocalPublication(sid string) *LocalTrackPublicatio
}
return nil
}

func (p *LocalParticipant) onTrackMuted(pub *LocalTrackPublication, muted bool) {
if muted {
p.Callback.OnTrackMuted(pub, p)
p.roomCallback.OnTrackMuted(pub, p)
} else {
p.Callback.OnTrackUnmuted(pub, p)
p.roomCallback.OnTrackUnmuted(pub, p)
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a chance of this racing and setting wrong callbacks?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, the callback is set in the CreateRoom function.

}
13 changes: 12 additions & 1 deletion publication.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,7 @@ type LocalTrackPublication struct {
simulcastTracks map[livekit.VideoQuality]*LocalSampleTrack
onRttUpdate func(uint32)
opts TrackPublicationOptions
onMuteChanged func(*LocalTrackPublication, bool)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we make this OnMuteChanged (upper case O) as it is set directly from the outside?

Copy link
Contributor Author

@cnderrauber cnderrauber Nov 20, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will cause the callback can be accessed by the user, so keep it lowercase to be access only by internal code

}

func NewLocalTrackPublication(kind TrackKind, track Track, opts TrackPublicationOptions, client *SignalClient) *LocalTrackPublication {
Expand Down Expand Up @@ -280,17 +281,27 @@ func (p *LocalTrackPublication) GetSimulcastTrack(quality livekit.VideoQuality)
}

func (p *LocalTrackPublication) SetMuted(muted bool) {
p.setMuted(muted, false)
}

func (p *LocalTrackPublication) setMuted(muted bool, byRemote bool) {
if p.isMuted.Swap(muted) == muted {
return
}

_ = p.client.SendMuteTrack(p.sid.Load(), muted)
if !byRemote {
_ = p.client.SendMuteTrack(p.sid.Load(), muted)
}
if track := p.track; track != nil {
switch t := track.(type) {
case *LocalSampleTrack:
t.setMuted(muted)
}
}

if p.onMuteChanged != nil {
p.onMuteChanged(p, muted)
}
}

func (p *LocalTrackPublication) addSimulcastTrack(st *LocalSampleTrack) {
Expand Down
6 changes: 3 additions & 3 deletions room.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ func CreateRoom(callback *RoomCallback) *Room {
engine.OnResuming = r.handleResuming
engine.OnResumed = r.handleResumed
engine.client.OnLocalTrackUnpublished = r.handleLocalTrackUnpublished
engine.client.OnTrackMuted = r.handleTrackMuted
engine.client.OnTrackRemoteMuted = r.handleTrackRemoteMuted

return r
}
Expand Down Expand Up @@ -484,12 +484,12 @@ func (r *Room) handleRoomUpdate(room *livekit.Room) {
go r.callback.OnRoomMetadataChanged(room.Metadata)
}

func (r *Room) handleTrackMuted(msg *livekit.MuteTrackRequest) {
func (r *Room) handleTrackRemoteMuted(msg *livekit.MuteTrackRequest) {
for _, pub := range r.LocalParticipant.Tracks() {
if pub.SID() == msg.Sid {
localPub := pub.(*LocalTrackPublication)
// TODO: pause sending data because it'll be dropped by SFU
localPub.SetMuted(msg.Muted)
localPub.setMuted(msg.Muted, true)
}
}
}
Expand Down
6 changes: 3 additions & 3 deletions signalclient.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ type SignalClient struct {
OnSpeakersChanged func([]*livekit.SpeakerInfo)
OnConnectionQuality func([]*livekit.ConnectionQualityInfo)
OnRoomUpdate func(room *livekit.Room)
OnTrackMuted func(request *livekit.MuteTrackRequest)
OnTrackRemoteMuted func(request *livekit.MuteTrackRequest)
OnLocalTrackUnpublished func(response *livekit.TrackUnpublishedResponse)
OnTokenRefresh func(refreshToken string)
OnLeave func(*livekit.LeaveRequest)
Expand Down Expand Up @@ -319,8 +319,8 @@ func (c *SignalClient) handleResponse(res *livekit.SignalResponse) {
c.OnLocalTrackPublished(msg.TrackPublished)
}
case *livekit.SignalResponse_Mute:
if c.OnTrackMuted != nil {
c.OnTrackMuted(msg.Mute)
if c.OnTrackRemoteMuted != nil {
c.OnTrackRemoteMuted(msg.Mute)
}
case *livekit.SignalResponse_ConnectionQuality:
if c.OnConnectionQuality != nil {
Expand Down
Loading