package localserver import ( "context" "encoding/json" "errors" "fmt" "io" "strconv" sync "sync" "sync/atomic" "time" "github.com/google/uuid" "github.com/pion/rtcp" "github.com/pion/webrtc/v3" ) const ( VIDEO_CHANNEL_ACCESS_DENIED ReqType = "video_channel_access_denied" VIDEO_CHANNEL_STOP_CALL ReqType = "video_channel_stop_call" VIDEO_CHANNEL_ACCESS_GRANTED ReqType = "video_channel_access_granted" VIDEO_CHANNEL_WEBRTC_OFFER ReqType = "video_channel_offer" VIDEO_CHANNEL_WEBRTC_ANSWER ReqType = "video_channel_answer" VIDEO_CHANNEL_WEBRTC_RENNEGOTIATION_OFFER ReqType = "video_channel_rennegotiation_offer" VIDEO_CHANNEL_WEBRTC_RENNEGOTIATION_ANSWER ReqType = "video_channel_rennegotiation_answer" VIDEO_CHANNEL_WEBRTC_COUNTER_OFFER ReqType = "video_channel_webrtc_counter_offer" VIDEO_CHANNEL_WEBRTC_CANDIDATE ReqType = "video_channel_webrtc_candidate" VIDEO_CHANNEL_REMOVE_VIDEO ReqType = "video_channel_remove_video" GET_VIDEO_CHANNEL_TRACKS ReqType = "video_channel_get_tracks" ) const ( VIDEO_CHANNEL_USER_VIDEO_STOP = "video_channel_user_video_stop" VIDEO_CHANNEL_USER_VIDEO_RESUME = "video_channel_user_video_resume" VIDEO_CHANNEL_USER_MUTE = "video_channel_user_mute" VIDEO_CHANNEL_USER_UNMUTE = "video_channel_user_unmute" VIDEO_CHANNEL_USER_SPEAKING = "video_channel_user_speaking" VIDEO_CHANNEL_USER_STOPPED_SPEAKING = "video_channel_user_stopped_speaking" ) type VideoChannel struct { ID string `json:"id"` Owner string `json:"owner"` ChannelType string `json:"channelType"` Members []string `json:"members"` CurrentMembersId []string `json:"currentMembersId"` CurrentMembers map[string]*VideoChannelMember `json:"currentMembers"` localSD map[string]*webrtc.SessionDescription `json:"-"` rtcPeerConnections map[string]*ZoneRTCPeerConnection `json:"-"` audioTransceiver map[string][]*PeerSender `json:"-"` videoTransceiver map[string][]*PeerSender `json:"-"` videoChannelDataChannels map[string]*DataChannel `json:"-"` pendingCandidates map[string][]*webrtc.ICECandidate `json:"-"` remoteTracks map[string][]*RemoteTrack `json:"-"` middlewares []interface{} `json:"-"` candidateFlag *uint32 `json:"-"` remoteTracksFlag *uint32 `json:"-"` rtcPeerConnectionMapFlag *uint32 `json:"-"` dataChannelMapFlag *uint32 `json:"-"` localSDMapFlag *uint32 `json:"-"` audioSenderFlag *uint32 `json:"-"` videoSenderFlag *uint32 `json:"-"` } type VideoChannelOnICECandidateFunc = func(string, string, *webrtc.ICECandidate) error func NewVideoChannel(id string, owner string, channelType string, members []string, currentMembersId []string, currentMembers map[string]*VideoChannelMember) (audioChannel *VideoChannel) { candidateFlag := uint32(0) remoteTracksFlag := uint32(0) rtcPeerConnectionMapFlag := uint32(0) dataChannelMapFlag := uint32(0) localSDMapFlag := uint32(0) audioSenderFlag := uint32(0) videoSenderFlag := uint32(0) audioChannel = &VideoChannel{ ID: id, Owner: owner, ChannelType: channelType, Members: members, CurrentMembersId: currentMembersId, CurrentMembers: currentMembers, localSD: make(map[string]*webrtc.SessionDescription), videoTransceiver: make(map[string][]*PeerSender), rtcPeerConnections: make(map[string]*ZoneRTCPeerConnection), audioTransceiver: make(map[string][]*PeerSender), videoChannelDataChannels: make(map[string]*DataChannel), pendingCandidates: make(map[string][]*webrtc.ICECandidate), remoteTracks: make(map[string][]*RemoteTrack), middlewares: make([]interface{}, 0), candidateFlag: &candidateFlag, remoteTracksFlag: &remoteTracksFlag, rtcPeerConnectionMapFlag: &rtcPeerConnectionMapFlag, dataChannelMapFlag: &dataChannelMapFlag, localSDMapFlag: &localSDMapFlag, audioSenderFlag: &audioSenderFlag, videoSenderFlag: &videoSenderFlag, } return } func (vc *VideoChannel) HandleOffer(ctx context.Context, channelId string, userId string, sdp string, hostId string, sendDCMessage SendDCMessageFunc, cb VideoChannelOnICECandidateFunc) (done chan struct{}, errCh chan error) { done, errCh = make(chan struct{}), make(chan error) go func() { peerConnection, err := vc.createPeerConnection(userId, vc.ID, webrtc.SDPTypeAnswer, cb, sendDCMessage) if err != nil { errCh <- err return } _ = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { vc.rtcPeerConnections[userId] = &ZoneRTCPeerConnection{ PeerConnection: peerConnection, makingOffer: false, makingOfferLock: &sync.Mutex{}, negotiate: vc.negotiate, } return }) offer := webrtc.SessionDescription{ Type: webrtc.SDPTypeOffer, SDP: sdp, } if err = peerConnection.SetRemoteDescription(offer); err != nil { errCh <- err return } rawAnswer, err := peerConnection.CreateAnswer(nil) if err != nil { errCh <- err return } if err = peerConnection.SetLocalDescription(rawAnswer); err != nil { errCh <- err return } _, _ = sendDCMessage(string(VIDEO_CHANNEL_WEBRTC_ANSWER), hostId, userId, map[string]interface{}{ "to": userId, "from": vc.ID, "channelId": channelId, "sdp": rawAnswer.SDP, }) done <- struct{}{} logger.Println("handle offer done") }() return } func (vc *VideoChannel) HandleCounterOffer(ctx context.Context, userId string, sendDCMessage SendDCMessageFunc) (err error) { // if err = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { // if _, ok := vc.rtcPeerConnections[userId]; !ok { // err = fmt.Errorf("no field corresponding peer connection for id %s", userId) // return // } // logger.Println("handling counter offer") // connection := vc.rtcPeerConnections[userId] // err = atomicallyExecute(vc.localSDMapFlag, func() (err error) { // err = connection.SetLocalDescription(*vc.localSD[userId]) // return // }) // return // }); err != nil { // return // } // _ = atomicallyExecute(vc.localSDMapFlag, func() (err error) { // delete(vc.localSD, userId) // return // }) if err = atomicallyExecute(vc.candidateFlag, func() (err error) { for _, candidate := range vc.pendingCandidates[userId] { logger.Println("sending candidate to", userId, candidate) d, e := sendDCMessage(string(VIDEO_CHANNEL_WEBRTC_CANDIDATE), "", userId, map[string]interface{}{ "from": vc.ID, "to": userId, "candidate": candidate.ToJSON().Candidate, "sdpMid": *candidate.ToJSON().SDPMid, "sdpMLineIndex": strconv.Itoa(int(*candidate.ToJSON().SDPMLineIndex)), }) select { case <-d: case err = <-e: return } } delete(vc.pendingCandidates, userId) return }); err != nil { return } return } func (vc *VideoChannel) HandleRennegotiationOffer(from string, sdp string, sendDCMessage SendDCMessageFunc) (err error) { err = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { if _, ok := vc.rtcPeerConnections[from]; !ok { err = fmt.Errorf("no corresponding peer connection for id %s", from) return } // vc.rtcPeerConnections[from].makingOfferLock.Lock() // if vc.rtcPeerConnections[from].makingOffer { // vc.rtcPeerConnections[from].makingOfferLock.Unlock() // return fmt.Errorf("already making an offer or state is stable") // } // vc.rtcPeerConnections[from].makingOfferLock.Unlock() if err = vc.rtcPeerConnections[from].SetRemoteDescription(webrtc.SessionDescription{SDP: sdp, Type: webrtc.SDPTypeOffer}); err != nil { return } localSd, err := vc.rtcPeerConnections[from].CreateAnswer(nil) if err != nil { return } if err = vc.rtcPeerConnections[from].SetLocalDescription(localSd); err != nil { return } d, e := sendDCMessage(string(VIDEO_CHANNEL_WEBRTC_RENNEGOTIATION_ANSWER), vc.ID, from, map[string]interface{}{ "from": vc.ID, "to": from, "sdp": localSd.SDP, }) select { case <-d: case err = <-e: } return }) return } func (vc *VideoChannel) HandleRennegotiationAnswer(from string, sdp string) (err error) { logger.Println("---------------------handling rennego answer") err = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { // vc.rtcPeerConnections[from].makingOfferLock.Lock() // if vc.rtcPeerConnections[from].makingOffer { // vc.rtcPeerConnections[from].makingOfferLock.Unlock() // return fmt.Errorf("already making an offer or state is stable") // } // vc.rtcPeerConnections[from].makingOfferLock.Unlock() // if _, ok := vc.rtcPeerConnections[from]; !ok { // err = fmt.Errorf("no corresponding peer connection for id %s", from) // return // } err = vc.rtcPeerConnections[from].SetRemoteDescription(webrtc.SessionDescription{SDP: sdp, Type: webrtc.SDPTypeAnswer}) return }) return } func (vc *VideoChannel) AddCandidate(candidate *webrtc.ICECandidateInit, from string) (err error) { logger.Println("adding ice candidate", candidate) err = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { if _, ok := vc.rtcPeerConnections[from]; ok && candidate != nil { err = vc.rtcPeerConnections[from].AddICECandidate(*candidate) } return }) return } func (vc *VideoChannel) createPeerConnection(target string, from string, peerType webrtc.SDPType, cb VideoChannelOnICECandidateFunc, sendDCMessage SendDCMessageFunc) (peerConnection *webrtc.PeerConnection, err error) { defer func() { if r := recover(); err != nil { logger.Printf("recover from panic : %v\n", r) } }() config := webrtc.Configuration{ ICEServers: []webrtc.ICEServer{ { URLs: []string{"stun:stun.l.google.com:19302", "stun:stunserver.org:3478"}, }, }, SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback, } peerConnection, err = webrtc.NewPeerConnection(config) if err != nil { return } logger.Println("---------------------------------------------------") if peerType == webrtc.SDPTypeAnswer { maxRetransmits := uint16(100) channel, err := peerConnection.CreateDataChannel("video-channel", &webrtc.DataChannelInit{ MaxRetransmits: &maxRetransmits, }) if err != nil { return nil, err } channel.OnOpen(func() { logger.Println("channel opened") if chanErr := channel.SendText("yooo man this is open"); chanErr != nil { logger.Println(chanErr) } }) channel.OnMessage(func(msg webrtc.DataChannelMessage) { var event CallEvent if err := json.Unmarshal(msg.Data, &event); err != nil { logger.Println(err) return } if e := vc.HandleDataChannelEvents(event.From, event.EventId, event.Payload); e != nil { logger.Println("*-------------- datachannel error: ", e) } }) logger.Println("new channel for target : ", target) _ = atomicallyExecute(vc.dataChannelMapFlag, func() (err error) { logger.Println(target) l := uint32(0) vc.videoChannelDataChannels[target] = &DataChannel{ DataChannel: channel, l: &l, } return }) } else { peerConnection.OnDataChannel(func(dc *webrtc.DataChannel) { _ = atomicallyExecute(vc.dataChannelMapFlag, func() (err error) { l := uint32(0) vc.videoChannelDataChannels[target] = &DataChannel{ DataChannel: dc, l: &l, } return }) dc.OnOpen(func() { logger.Printf("got a new open datachannel %s\n", dc.Label()) }) dc.OnMessage(func(msg webrtc.DataChannelMessage) { var event CallEvent if err := json.Unmarshal(msg.Data, &event); err != nil { logger.Println(err) return } if e := vc.HandleDataChannelEvents(event.From, event.EventId, event.Payload); e != nil { logger.Println("*-------------- datachannel error: ", e) } }) }) } err = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { err = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { logger.Println("------------------", vc.CurrentMembersId) for _, id := range vc.CurrentMembersId { logger.Println(id) if id != target { if _, ok := vc.remoteTracks[id]; !ok { continue } for _, track := range vc.remoteTracks[id] { transceiver, err := peerConnection.AddTransceiverFromKind(track.Track.Kind(), webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}) if err != nil { logger.Println("add track error") continue } if err := transceiver.Sender().ReplaceTrack(track.Track); err != nil { logger.Println("add track error") continue } if track.Track.Kind() == webrtc.RTPCodecTypeVideo { _ = atomicallyExecute(vc.videoSenderFlag, func() (err error) { if len(vc.videoTransceiver) == 0 { vc.videoTransceiver[id] = []*PeerSender{{ID: target, Transceiver: transceiver}} } else { vc.videoTransceiver[id] = append(vc.videoTransceiver[id], &PeerSender{ID: target, Transceiver: transceiver}) } return }) } else if track.Track.Kind() == webrtc.RTPCodecTypeAudio { _ = atomicallyExecute(vc.audioSenderFlag, func() (err error) { if len(vc.audioTransceiver) == 0 { vc.audioTransceiver[id] = []*PeerSender{{ID: target, Transceiver: transceiver}} } else { vc.audioTransceiver[id] = append(vc.audioTransceiver[id], &PeerSender{ID: target, Transceiver: transceiver}) } return }) } logger.Println("track added", track) } } } return }) return }) peerConnection.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) { if pcs == webrtc.PeerConnectionStateClosed || pcs == webrtc.PeerConnectionStateDisconnected || pcs == webrtc.PeerConnectionStateFailed { logger.Println(pcs) //vc.HandleLeavingMember(target, squadId) } }) peerConnection.OnICEConnectionStateChange(func(is webrtc.ICEConnectionState) { logger.Printf("ICE connection state has changed %s\n", is.String()) if is == webrtc.ICEConnectionStateDisconnected || is == webrtc.ICEConnectionStateFailed { logger.Println(is) } }) peerConnection.OnTrack(func(tr *webrtc.TrackRemote, r *webrtc.RTPReceiver) { logger.Println("got new track") defer func() { if stopErr := r.Stop(); stopErr != nil { logger.Println(stopErr) } }() go func() { ticker := time.NewTicker(1500 * time.Millisecond) for range ticker.C { if rtcpSendErr := peerConnection.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(tr.SSRC())}}); rtcpSendErr != nil { logger.Println(rtcpSendErr) break } } }() uniqId := uuid.New() i := fmt.Sprintf("%s/%s", target, uniqId.String()) logger.Println("*************************----------------", i, "-----------------------***************") localTrack, newTrackErr := webrtc.NewTrackLocalStaticRTP(tr.Codec().RTPCodecCapability, i, i) if newTrackErr != nil { return } logger.Println(localTrack) rtpbuf := make([]byte, 1400) flag := int32(0) remote := &RemoteTrack{ID: target, Track: localTrack, rdv: &flag} _ = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { if len(vc.remoteTracks[target]) == 0 { vc.remoteTracks[target] = []*RemoteTrack{remote} } else { vc.remoteTracks[target] = append(vc.remoteTracks[target], remote) } index := len(vc.remoteTracks[target]) logger.Println(index, vc.remoteTracks) return }) _ = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { for _, id := range vc.CurrentMembersId { if id != target { if _, ok := vc.rtcPeerConnections[id]; !ok { continue } connection := vc.rtcPeerConnections[id] transceiver, tranceiverErr := connection.AddTransceiverFromKind(localTrack.Kind(), webrtc.RTPTransceiverInit{Direction: webrtc.RTPTransceiverDirectionSendonly}) if tranceiverErr != nil { logger.Println(tranceiverErr) continue } if replaceTrackErr := transceiver.Sender().ReplaceTrack(localTrack); replaceTrackErr != nil { logger.Println(replaceTrackErr) continue } go func() { rtcpBuf := make([]byte, 1500) for { if _, _, rtcpErr := transceiver.Sender().Read(rtcpBuf); rtcpErr != nil { return } } }() if localTrack.Kind() == webrtc.RTPCodecTypeAudio { _ = atomicallyExecute(vc.audioSenderFlag, func() (err error) { if len(vc.audioTransceiver) == 0 { vc.audioTransceiver[target] = []*PeerSender{{ID: id, Transceiver: transceiver}} } else { vc.audioTransceiver[target] = append(vc.audioTransceiver[target], &PeerSender{ID: id, Transceiver: transceiver}) } return }) } else if localTrack.Kind() == webrtc.RTPCodecTypeVideo { _ = atomicallyExecute(vc.videoSenderFlag, func() (err error) { if len(vc.videoTransceiver) == 0 { vc.videoTransceiver[target] = []*PeerSender{{ID: id, Transceiver: transceiver}} } else { vc.videoTransceiver[target] = append(vc.videoTransceiver[target], &PeerSender{ID: id, Transceiver: transceiver}) } return }) } go func() { <-time.After(time.Second * 1) connection.negotiate(id, sendDCMessage) }() } } return }) d := make(chan struct{}) go func() { for { i, _, readErr := tr.Read(rtpbuf) if readErr != nil { logger.Println(readErr) break } f := atomic.LoadInt32(remote.rdv) if f == 0 { if _, writeErr := localTrack.Write(rtpbuf[:i]); writeErr != nil && !errors.Is(writeErr, io.ErrClosedPipe) { logger.Println(writeErr) break } else { _ = rtpbuf[:i] } } } d <- struct{}{} }() <-d }) peerConnection.OnICECandidate(func(i *webrtc.ICECandidate) { if i == nil { return } _ = atomicallyExecute(vc.candidateFlag, func() (err error) { desc := peerConnection.RemoteDescription() if desc == nil { logger.Println("generated candidate appended to list : ", i) vc.pendingCandidates[target] = append(vc.pendingCandidates[target], i) } else { logger.Println("generated candidate : ", i) if iceCandidateErr := cb(from, target, i); iceCandidateErr != nil { logger.Println(iceCandidateErr) } } return }) }) peerConnection.OnNegotiationNeeded(func() { logger.Println("---------------- rennego is needed -----------") // _ = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { // for _, id := range vc.CurrentMembersId { // logger.Println("----------------- sending renego to peer with id", id) // if _, ok := vc.rtcPeerConnections[id]; !ok { // continue // } // if peerConnection.SignalingState() == webrtc.SignalingStateStable { // localSd, localSdErr := peerConnection.CreateOffer(nil) // if localSdErr != nil { // logger.Println(localSdErr) // return localSdErr // } // if err = peerConnection.SetLocalDescription(localSd); err != nil { // logger.Println(err) // return // } // d, e := sendDCMessage(string(VIDEO_CHANNEL_WEBRTC_RENNEGOTIATION_OFFER), vc.ID, id, map[string]interface{}{ // "from": vc.ID, // "to": id, // "sdp": localSd.SDP, // }) // select { // case <-d: // case err = <-e: // logger.Println(err) // } // } // } // return // }) }) return } func (vc *VideoChannel) HandleLeavingMember(id string) { if err := atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { if _, ok := vc.rtcPeerConnections[id]; !ok { err = fmt.Errorf("no corresponding peerconnection for audio channel leaving member") } return }); err != nil { logger.Println(err) } else { defer func() { _ = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { if _, ok := vc.rtcPeerConnections[id]; ok { if closeErr := vc.rtcPeerConnections[id].Close(); closeErr != nil { err = closeErr logger.Println("peer connection close error", closeErr) } } delete(vc.rtcPeerConnections, id) return }) }() } logger.Printf("peer %s is leaving the squad\n", id) _ = atomicallyExecute(vc.dataChannelMapFlag, func() (err error) { if _, ok := vc.videoChannelDataChannels[id]; ok { vc.videoChannelDataChannels[id].DataChannel.Close() } delete(vc.videoChannelDataChannels, id) return }) _ = atomicallyExecute(vc.localSDMapFlag, func() (err error) { delete(vc.localSD, id) return }) _ = atomicallyExecute(vc.candidateFlag, func() (err error) { delete(vc.pendingCandidates, id) return }) _ = atomicallyExecute(vc.audioSenderFlag, func() (err error) { for peerId, peerSender := range vc.audioTransceiver { if peerId != id { logger.Println("senders", peerSender) c := 0 for i, sender := range peerSender { if sender.ID == id { if senderErr := sender.Transceiver.Sender().Stop(); senderErr != nil { logger.Println(senderErr) } if transceiverErr := sender.Transceiver.Stop(); transceiverErr != nil { logger.Println("transceiverErr occured with video", transceiverErr) } peerSender[len(peerSender)-i-1], peerSender[i] = peerSender[i], peerSender[len(peerSender)-i-1] c++ } } vc.audioTransceiver[peerId] = vc.audioTransceiver[peerId][:len(peerSender)-(c)] logger.Println(vc.audioTransceiver[peerId]) } } for _, transceiver := range vc.audioTransceiver[id] { if senderErr := transceiver.Transceiver.Sender().Stop(); senderErr != nil { logger.Println(senderErr) } if stopErr := transceiver.Transceiver.Stop(); stopErr != nil { logger.Println("transceiver audio stop error", stopErr) } } delete(vc.audioTransceiver, id) return }) _ = atomicallyExecute(vc.videoSenderFlag, func() (err error) { for peerId, peerSender := range vc.videoTransceiver { if peerId != id { logger.Println("senders", peerSender) c := 0 for i, sender := range peerSender { if sender.ID == id { if senderErr := sender.Transceiver.Sender().Stop(); senderErr != nil { logger.Println(senderErr) } if transceiverErr := sender.Transceiver.Stop(); transceiverErr != nil { logger.Println("transceiverErr occured with video", transceiverErr) } peerSender[len(peerSender)-i-1], peerSender[i] = peerSender[i], peerSender[len(peerSender)-i-1] c++ } } vc.videoTransceiver[peerId] = vc.videoTransceiver[peerId][:len(peerSender)-(c)] logger.Println(vc.videoTransceiver[peerId]) } } for _, transceiver := range vc.videoTransceiver[id] { if senderErr := transceiver.Transceiver.Sender().Stop(); senderErr != nil { logger.Println(senderErr) } if stopErr := transceiver.Transceiver.Stop(); stopErr != nil { logger.Println("transceiver video stop error", stopErr) } } delete(vc.videoTransceiver, id) return }) _ = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { delete(vc.remoteTracks, id) return }) } func (vc *VideoChannel) negotiate(target string, sendDCMessage SendDCMessageFunc) { logger.Println("------------------negotiate is called") _ = atomicallyExecute(vc.rtcPeerConnectionMapFlag, func() (err error) { if _, ok := vc.rtcPeerConnections[target]; !ok { return } // vc.rtcPeerConnections[target].makingOfferLock.Lock() // vc.rtcPeerConnections[target].makingOffer = true // vc.rtcPeerConnections[target].makingOfferLock.Unlock() defer func() { // vc.rtcPeerConnections[target].makingOfferLock.Lock() // vc.rtcPeerConnections[target].makingOffer = false // vc.rtcPeerConnections[target].makingOfferLock.Unlock() }() for _, id := range vc.CurrentMembersId { logger.Println("----------------- sending renego to peer with id", id) if _, ok := vc.rtcPeerConnections[id]; !ok { continue } connection := vc.rtcPeerConnections[id] if connection.SignalingState() == webrtc.SignalingStateStable { localSd, err := connection.CreateOffer(nil) if err != nil { logger.Println(err) return err } if err = connection.SetLocalDescription(localSd); err != nil { logger.Println(err) return err } d, e := sendDCMessage(string(VIDEO_CHANNEL_WEBRTC_RENNEGOTIATION_OFFER), vc.ID, id, map[string]interface{}{ "from": vc.ID, "to": id, "sdp": localSd.SDP, }) select { case <-d: case err = <-e: logger.Println(err) } } } return }) } func (vc *VideoChannel) broadcastDatachannelMessage(from string, eventId string, payload map[string]interface{}) (done chan struct{}, errCh chan error) { done, errCh = make(chan struct{}), make(chan error) go func() { bs, jsonErr := json.Marshal(&ZoneResponse{ Type: eventId, From: vc.ID, Payload: payload, }) if jsonErr != nil { errCh <- jsonErr return } if err := atomicallyExecute(vc.dataChannelMapFlag, func() (err error) { for id, dc := range vc.videoChannelDataChannels { if from != id { if err = dc.DataChannel.SendText(string(bs)); err != nil { return } } } return }); err != nil { errCh <- err } done <- struct{}{} }() return } func (vc *VideoChannel) HandleDataChannelEvents(from string, eventId string, payload map[string]interface{}) (err error) { switch eventId { case VIDEO_CHANNEL_USER_VIDEO_STOP: if err = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { if _, ok := vc.remoteTracks[from]; !ok { err = fmt.Errorf("no corresponding remote tracks entry for id %s", from) return } for _, track := range vc.remoteTracks[from] { if track.Track.Kind() == webrtc.RTPCodecTypeVideo { atomic.SwapInt32(track.rdv, 1) } } return }); err != nil { return } done, errCh := vc.broadcastDatachannelMessage(from, VIDEO_CHANNEL_USER_VIDEO_STOP, map[string]interface{}{ "userId": from, }) select { case <-done: case err = <-errCh: } case VIDEO_CHANNEL_USER_VIDEO_RESUME: if err = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { if _, ok := vc.remoteTracks[from]; !ok { err = fmt.Errorf("no corresponding remote tracks entry for id %s", from) return } for _, track := range vc.remoteTracks[from] { if track.Track.Kind() == webrtc.RTPCodecTypeVideo { atomic.SwapInt32(track.rdv, 0) } } return }); err != nil { return } done, errCh := vc.broadcastDatachannelMessage(from, VIDEO_CHANNEL_USER_VIDEO_RESUME, map[string]interface{}{ "userId": from, }) select { case <-done: case err = <-errCh: } case VIDEO_CHANNEL_USER_MUTE: if err = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { if _, ok := vc.remoteTracks[from]; !ok { err = fmt.Errorf("no corresponding remote tracks entry for id %s", from) return } for _, track := range vc.remoteTracks[from] { if track.Track.Kind() == webrtc.RTPCodecTypeAudio { atomic.SwapInt32(track.rdv, 1) } } return }); err != nil { return } done, errCh := vc.broadcastDatachannelMessage(from, VIDEO_CHANNEL_USER_MUTE, map[string]interface{}{ "userId": from, }) select { case <-done: case err = <-errCh: } case VIDEO_CHANNEL_USER_UNMUTE: if err = atomicallyExecute(vc.remoteTracksFlag, func() (err error) { if _, ok := vc.remoteTracks[from]; !ok { err = fmt.Errorf("no corresponding remote tracks entry for id %s", from) return } for _, track := range vc.remoteTracks[from] { if track.Track.Kind() == webrtc.RTPCodecTypeAudio { atomic.SwapInt32(track.rdv, 0) } } return }); err != nil { return } done, errCh := vc.broadcastDatachannelMessage(from, VIDEO_CHANNEL_USER_UNMUTE, map[string]interface{}{ "userId": from, }) select { case <-done: case err = <-errCh: } case VIDEO_CHANNEL_USER_SPEAKING: done, errCh := vc.broadcastDatachannelMessage(from, VIDEO_CHANNEL_USER_SPEAKING, map[string]interface{}{ "userId": from, }) select { case <-done: case err = <-errCh: } case VIDEO_CHANNEL_USER_STOPPED_SPEAKING: done, errCh := vc.broadcastDatachannelMessage(from, VIDEO_CHANNEL_USER_STOPPED_SPEAKING, map[string]interface{}{ "userId": from, }) select { case <-done: case err = <-errCh: } } return }