diff --git a/pkg/rtc/supervisor/participant_supervisor.go b/pkg/rtc/supervisor/participant_supervisor.go index 687c34d21..5103fe82c 100644 --- a/pkg/rtc/supervisor/participant_supervisor.go +++ b/pkg/rtc/supervisor/participant_supervisor.go @@ -51,7 +51,7 @@ func (p *ParticipantSupervisor) AddPublication(trackID livekit.TrackID) { pm = NewPublicationMonitor(PublicationMonitorParams{TrackID: trackID, Logger: p.params.Logger}) p.publications[trackID] = pm } - pm.(*PublicationMonitor).AddPending() + pm.PostEvent(types.OperationMonitorEventAddPendingPublication, nil) p.lock.Unlock() } @@ -59,7 +59,7 @@ func (p *ParticipantSupervisor) SetPublicationMute(trackID livekit.TrackID, isMu p.lock.Lock() pm, ok := p.publications[trackID] if ok { - pm.(*PublicationMonitor).SetMute(isMuted) + pm.PostEvent(types.OperationMonitorEventSetPublicationMute, isMuted) } p.lock.Unlock() } @@ -68,7 +68,7 @@ func (p *ParticipantSupervisor) SetPublishedTrack(trackID livekit.TrackID, pubTr p.lock.RLock() pm, ok := p.publications[trackID] if ok { - pm.(*PublicationMonitor).SetPublishedTrack(pubTrack) + pm.PostEvent(types.OperationMonitorEventSetPublishedTrack, pubTrack) } p.lock.RUnlock() } @@ -77,7 +77,7 @@ func (p *ParticipantSupervisor) ClearPublishedTrack(trackID livekit.TrackID, pub p.lock.RLock() pm, ok := p.publications[trackID] if ok { - pm.(*PublicationMonitor).ClearPublishedTrack(pubTrack) + pm.PostEvent(types.OperationMonitorEventClearPublishedTrack, pubTrack) } p.lock.RUnlock() } @@ -89,7 +89,7 @@ func (p *ParticipantSupervisor) UpdateSubscription(trackID livekit.TrackID, isSu sm = NewSubscriptionMonitor(SubscriptionMonitorParams{TrackID: trackID, Logger: p.params.Logger}) p.subscriptions[trackID] = sm } - sm.(*SubscriptionMonitor).UpdateSubscription(isSubscribed) + sm.PostEvent(types.OperationMonitorEventUpdateSubscription, isSubscribed) p.lock.Unlock() } @@ -97,7 +97,7 @@ func (p *ParticipantSupervisor) SetSubscribedTrack(trackID livekit.TrackID, subT p.lock.RLock() sm, ok := p.subscriptions[trackID] if ok { - sm.(*SubscriptionMonitor).SetSubscribedTrack(subTrack) + sm.PostEvent(types.OperationMonitorEventSetSubscribedTrack, subTrack) } p.lock.RUnlock() } @@ -106,7 +106,7 @@ func (p *ParticipantSupervisor) ClearSubscribedTrack(trackID livekit.TrackID, su p.lock.RLock() sm, ok := p.subscriptions[trackID] if ok { - sm.(*SubscriptionMonitor).ClearSubscribedTrack(subTrack) + sm.PostEvent(types.OperationMonitorEventClearSubscribedTrack, subTrack) } p.lock.RUnlock() } diff --git a/pkg/rtc/supervisor/publication_monitor.go b/pkg/rtc/supervisor/publication_monitor.go index 2f99b0470..557323a2a 100644 --- a/pkg/rtc/supervisor/publication_monitor.go +++ b/pkg/rtc/supervisor/publication_monitor.go @@ -47,7 +47,20 @@ func NewPublicationMonitor(params PublicationMonitorParams) *PublicationMonitor return p } -func (p *PublicationMonitor) AddPending() { +func (p *PublicationMonitor) PostEvent(ome types.OperationMonitorEvent, omd types.OperationMonitorData) { + switch ome { + case types.OperationMonitorEventAddPendingPublication: + p.addPending() + case types.OperationMonitorEventSetPublicationMute: + p.setMute(omd.(bool)) + case types.OperationMonitorEventSetPublishedTrack: + p.setPublishedTrack(omd.(types.LocalMediaTrack)) + case types.OperationMonitorEventClearPublishedTrack: + p.clearPublishedTrack(omd.(types.LocalMediaTrack)) + } +} + +func (p *PublicationMonitor) addPending() { p.lock.Lock() p.desiredPublishes.PushBack( &publish{ @@ -65,7 +78,7 @@ func (p *PublicationMonitor) AddPending() { p.lock.Unlock() } -func (p *PublicationMonitor) SetMute(isMuted bool) { +func (p *PublicationMonitor) setMute(isMuted bool) { p.lock.Lock() p.isMuted = isMuted if !p.isMuted { @@ -74,14 +87,14 @@ func (p *PublicationMonitor) SetMute(isMuted bool) { p.lock.Unlock() } -func (p *PublicationMonitor) SetPublishedTrack(pubTrack types.LocalMediaTrack) { +func (p *PublicationMonitor) setPublishedTrack(pubTrack types.LocalMediaTrack) { p.lock.Lock() p.publishedTrack = pubTrack p.update() p.lock.Unlock() } -func (p *PublicationMonitor) ClearPublishedTrack(pubTrack types.LocalMediaTrack) { +func (p *PublicationMonitor) clearPublishedTrack(pubTrack types.LocalMediaTrack) { p.lock.Lock() if p.publishedTrack == pubTrack { p.publishedTrack = nil diff --git a/pkg/rtc/supervisor/subscription_monitor.go b/pkg/rtc/supervisor/subscription_monitor.go index 35fb6c872..0187abe5f 100644 --- a/pkg/rtc/supervisor/subscription_monitor.go +++ b/pkg/rtc/supervisor/subscription_monitor.go @@ -46,7 +46,18 @@ func NewSubscriptionMonitor(params SubscriptionMonitorParams) *SubscriptionMonit return s } -func (s *SubscriptionMonitor) UpdateSubscription(isSubscribed bool) { +func (s *SubscriptionMonitor) PostEvent(ome types.OperationMonitorEvent, omd types.OperationMonitorData) { + switch ome { + case types.OperationMonitorEventUpdateSubscription: + s.updateSubscription(omd.(bool)) + case types.OperationMonitorEventSetSubscribedTrack: + s.setSubscribedTrack(omd.(types.SubscribedTrack)) + case types.OperationMonitorEventClearSubscribedTrack: + s.clearSubscribedTrack(omd.(types.SubscribedTrack)) + } +} + +func (s *SubscriptionMonitor) updateSubscription(isSubscribed bool) { s.lock.Lock() s.desiredTransitions.PushBack( &transition{ @@ -58,14 +69,14 @@ func (s *SubscriptionMonitor) UpdateSubscription(isSubscribed bool) { s.lock.Unlock() } -func (s *SubscriptionMonitor) SetSubscribedTrack(subTrack types.SubscribedTrack) { +func (s *SubscriptionMonitor) setSubscribedTrack(subTrack types.SubscribedTrack) { s.lock.Lock() s.subscribedTrack = subTrack s.update() s.lock.Unlock() } -func (s *SubscriptionMonitor) ClearSubscribedTrack(subTrack types.SubscribedTrack) { +func (s *SubscriptionMonitor) clearSubscribedTrack(subTrack types.SubscribedTrack) { s.lock.Lock() if s.subscribedTrack == subTrack { s.subscribedTrack = nil diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 32bc5d389..e0900ba0f 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -420,7 +420,46 @@ type SubscribedTrack interface { UpdateVideoLayer() } +// +// Supervisor/operation monitor related definitions +// +type OperationMonitorEvent int + +const ( + OperationMonitorEventUpdateSubscription OperationMonitorEvent = iota + OperationMonitorEventSetSubscribedTrack + OperationMonitorEventClearSubscribedTrack + OperationMonitorEventAddPendingPublication + OperationMonitorEventSetPublicationMute + OperationMonitorEventSetPublishedTrack + OperationMonitorEventClearPublishedTrack +) + +func (o OperationMonitorEvent) String() string { + switch o { + case OperationMonitorEventUpdateSubscription: + return "UPDATE_SUBSCRIPTION" + case OperationMonitorEventSetSubscribedTrack: + return "SET_SUBSCRIBED_TRACK" + case OperationMonitorEventClearSubscribedTrack: + return "CLEAR_SUBSCRIBED_TRACK" + case OperationMonitorEventAddPendingPublication: + return "ADD_PENDING_PUBLICATION" + case OperationMonitorEventSetPublicationMute: + return "SET_PUBLICATION_MUTE" + case OperationMonitorEventSetPublishedTrack: + return "SET_PUBLISHED_TRACK" + case OperationMonitorEventClearPublishedTrack: + return "CLEAR_PUBLISHED_TRACK" + default: + return fmt.Sprintf("%d", int(o)) + } +} + +type OperationMonitorData interface{} + type OperationMonitor interface { + PostEvent(ome OperationMonitorEvent, omd OperationMonitorData) Check() error IsIdle() bool }