package localserver import ( "encoding/json" "fmt" "sync/atomic" "github.com/pion/webrtc/v3" ) const ( SPEAKING = "speaking" STOP_SPEAKING = "stop_speaking" MUTE = "mute" UNMUTE = "unmute" ) type WebrtcCallSoundManager struct{} func NewWebrtcCallSoundManager() *WebrtcCallSoundManager { return new(WebrtcCallSoundManager) } func (w *WebrtcCallSoundManager) HandleCallEvent(from string, squadId string, eventId string, payload map[string]interface{}, data []byte, manager *WebRTCCallManager) (err error) { done, errCh := make(chan struct{}), make(chan error) go func() { logger.Println("got an event in call sound manager", from, eventId, payload) switch eventId { case UNMUTE: if e := w.unmute(from, squadId, manager); e != nil { errCh <- e return } case MUTE: if e := w.mute(from, squadId, manager); e != nil { errCh <- e return } case SPEAKING: if e := w.sendSpeakingEvent(from, squadId, manager); e != nil { errCh <- e return } case STOP_SPEAKING: if e := w.sendStopSpeakingEvent(from, squadId, manager); e != nil { errCh <- e return } } done <- struct{}{} }() select { case <-done: return nil case err = <-errCh: return } } func (w *WebrtcCallSoundManager) unmute(peerId string, squadId string, manager *WebRTCCallManager) (err error) { if _, ok := manager.Squads[squadId]; !ok { err = fmt.Errorf("no correponding squad found") return } logger.Println("sending unnmute event", peerId) manager.RemoteTracksMux.RLock() for _, v := range manager.RemoteTracks[peerId] { if v.Track.Kind() == webrtc.RTPCodecTypeAudio { atomic.SwapInt32(v.rdv, 0) } } manager.RemoteTracksMux.RUnlock() manager.SquadMapMux.Lock() defer manager.SquadMapMux.Unlock() for _, member := range manager.Squads[squadId].Members { manager.DataChannelMapMux.Lock() if _, ok := manager.DataChannels[member]; ok && member != peerId { bs, marshalErr := json.Marshal(map[string]interface{}{ "type": UNMUTE, "from": peerId, "payload": map[string]interface{}{}, }) if marshalErr != nil { logger.Println(err) manager.DataChannelMapMux.Unlock() continue } lock: for { if atomic.CompareAndSwapInt32(manager.DataChannels[member].l, 0, 1) { defer atomic.SwapInt32(manager.DataChannels[member].l, 0) if sendErr := manager.DataChannels[member].DataChannel.SendText(string(bs)); sendErr != nil { logger.Println(sendErr) } break lock } else { continue } } } manager.DataChannelMapMux.Unlock() } return } func (w *WebrtcCallSoundManager) mute(peerId string, squadId string, manager *WebRTCCallManager) (err error) { if _, ok := manager.Squads[squadId]; !ok { err = fmt.Errorf("no correponding squad found") return } logger.Println("sending mute event", peerId) manager.RemoteTracksMux.RLock() for _, v := range manager.RemoteTracks[peerId] { if v.Track.Kind() == webrtc.RTPCodecTypeAudio { atomic.SwapInt32(v.rdv, 1) } } manager.RemoteTracksMux.RUnlock() manager.SquadMapMux.Lock() defer manager.SquadMapMux.Unlock() for _, member := range manager.Squads[squadId].Members { manager.DataChannelMapMux.Lock() if _, ok := manager.DataChannels[member]; ok && member != peerId { bs, marshalErr := json.Marshal(map[string]interface{}{ "type": MUTE, "from": peerId, "payload": map[string]interface{}{}, }) if marshalErr != nil { logger.Println(err) manager.DataChannelMapMux.Unlock() continue } lock: for { if atomic.CompareAndSwapInt32(manager.DataChannels[member].l, 0, 1) { defer atomic.SwapInt32(manager.DataChannels[member].l, 0) if sendErr := manager.DataChannels[member].DataChannel.SendText(string(bs)); sendErr != nil { logger.Println(sendErr) } break lock } else { continue } } } manager.DataChannelMapMux.Unlock() } return } func (w *WebrtcCallSoundManager) sendSpeakingEvent(peerId string, squadId string, manager *WebRTCCallManager) (err error) { if _, ok := manager.Squads[squadId]; !ok { err = fmt.Errorf("no correponding squad found") return } logger.Println("sending speaking event", peerId) manager.SquadMapMux.Lock() defer manager.SquadMapMux.Unlock() for _, member := range manager.Squads[squadId].Members { manager.DataChannelMapMux.Lock() if _, ok := manager.DataChannels[member]; ok && member != peerId { bs, marshalErr := json.Marshal(map[string]interface{}{ "type": SPEAKING, "from": peerId, "payload": map[string]interface{}{ "userId": peerId, "speaking": true, }, }) if marshalErr != nil { logger.Println(err) manager.DataChannelMapMux.Unlock() continue } lock: for { if atomic.CompareAndSwapInt32(manager.DataChannels[member].l, 0, 1) { defer atomic.SwapInt32(manager.DataChannels[member].l, 0) if sendErr := manager.DataChannels[member].DataChannel.SendText(string(bs)); sendErr != nil { logger.Println(sendErr) } break lock } else { continue } } } manager.DataChannelMapMux.Unlock() } return } func (w *WebrtcCallSoundManager) sendStopSpeakingEvent(peerId string, squadId string, manager *WebRTCCallManager) (err error) { if _, ok := manager.Squads[squadId]; !ok { err = fmt.Errorf("no correponding squad found") return } logger.Println("sending stop speaking event", peerId) manager.SquadMapMux.Lock() defer manager.SquadMapMux.Unlock() for _, member := range manager.Squads[squadId].Members { manager.DataChannelMapMux.Lock() if _, ok := manager.DataChannels[member]; ok && member != peerId { bs, marshalErr := json.Marshal(map[string]interface{}{ "type": STOP_SPEAKING, "from": peerId, "payload": map[string]interface{}{ "userId": peerId, "speaking": false, }, }) if marshalErr != nil { logger.Println(err) manager.DataChannelMapMux.Unlock() continue } lock: for { if atomic.CompareAndSwapInt32(manager.DataChannels[member].l, 0, 1) { defer atomic.SwapInt32(manager.DataChannels[member].l, 0) if sendErr := manager.DataChannels[member].DataChannel.SendText(string(bs)); sendErr != nil { logger.Println(sendErr) } break lock } else { continue } } } manager.DataChannelMapMux.Unlock() } return }