Merge pull request #166 from kerberos-io/fix/candidategather-extended-buffer

Fix ice candidate gather and extended candidate buffer

Dockerfile

@@ -61,7 +61,7 @@ RUN /dist/agent/main version
 ###############################################
 # Build Bento4 -> we want fragmented mp4 files
 
-ENV BENTO4_VERSION 1.6.0-639
+ENV BENTO4_VERSION 1.6.0-641
 RUN cd /tmp && git clone https://github.com/axiomatic-systems/Bento4 && cd Bento4 && \
 	git checkout tags/v${BENTO4_VERSION} && \
 	cd Build && \

machinery/src/utils/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
-const VERSION = "3.3.3"
+const VERSION = "3.3.5"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 

machinery/src/webrtc/main.go

@@ -25,7 +25,6 @@ var (
 	CandidateArrays     map[string](chan string)
 	peerConnectionCount int64
 	peerConnections     map[string]*pionWebRTC.PeerConnection
-	//encoder             *ffmpeg.VideoEncoder
 )
 
 type WebRTC struct {
@@ -38,24 +37,6 @@ type WebRTC struct {
 	PacketsCount          chan int
 }
 
-// No longer used, is for transcoding, might comeback on this!
-/*func init() {
-	// Encoder is created for once and for all.
-	var err error
-	encoder, err = ffmpeg.NewVideoEncoderByCodecType(av.H264)
-	if err != nil {
-		return
-	}
-	if encoder == nil {
-		err = fmt.Errorf("Video encoder not found")
-		return
-	}
-	encoder.SetFramerate(30, 1)
-	encoder.SetPixelFormat(av.I420)
-	encoder.SetBitrate(1000000) // 1MB
-	encoder.SetGopSize(30 / 1)  // 1s
-}*/
-
 func CreateWebRTC(name string, stunServers []string, turnServers []string, turnServersUsername string, turnServersCredential string) *WebRTC {
 	return &WebRTC{
 		Name:                  name,
@@ -64,7 +45,6 @@ func CreateWebRTC(name string, stunServers []string, turnServers []string, turnS
 		TurnServersUsername:   turnServersUsername,
 		TurnServersCredential: turnServersCredential,
 		Timer:                 time.NewTimer(time.Second * 10),
-		PacketsCount:          make(chan int),
 	}
 }
 
@@ -90,7 +70,7 @@ func RegisterCandidates(key string, candidate models.ReceiveHDCandidatesPayload)
 	CandidatesMutex.Lock()
 	_, ok := CandidateArrays[key]
 	if !ok {
-		CandidateArrays[key] = make(chan string)
+		CandidateArrays[key] = make(chan string, 100)
 	}
 	log.Log.Info("webrtc.main.HandleReceiveHDCandidates(): " + candidate.Candidate)
 	select {
@@ -115,7 +95,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 	CandidatesMutex.Lock()
 	_, ok := CandidateArrays[sessionKey]
 	if !ok {
-		CandidateArrays[sessionKey] = make(chan string)
+		CandidateArrays[sessionKey] = make(chan string, 100)
 	}
 	CandidatesMutex.Unlock()
 
@@ -167,41 +147,53 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding audio track: " + err.Error())
 			}
 
-			peerConnection.OnICEConnectionStateChange(func(connectionState pionWebRTC.ICEConnectionState) {
-				if connectionState == pionWebRTC.ICEConnectionStateDisconnected {
-					atomic.AddInt64(&peerConnectionCount, -1)
-
+			peerConnection.OnConnectionStateChange(func(connectionState pionWebRTC.PeerConnectionState) {
+				if connectionState == pionWebRTC.PeerConnectionStateDisconnected || connectionState == pionWebRTC.PeerConnectionStateClosed {
 					// Set lock
 					CandidatesMutex.Lock()
-					peerConnections[handshake.SessionID] = nil
+					atomic.AddInt64(&peerConnectionCount, -1)
 					_, ok := CandidateArrays[sessionKey]
 					if ok {
 						close(CandidateArrays[sessionKey])
+						delete(CandidateArrays, sessionKey)
 					}
-					CandidatesMutex.Unlock()
-
-					close(w.PacketsCount)
+					// Not really needed.
+					//senders := peerConnection.GetSenders()
+					//for _, sender := range senders {
+					//	if err := peerConnection.RemoveTrack(sender); err != nil {
+					//		log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while removing track: " + err.Error())
+					//	}
+					//}
 					if err := peerConnection.Close(); err != nil {
 						log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while closing peer connection: " + err.Error())
 					}
-				} else if connectionState == pionWebRTC.ICEConnectionStateConnected {
+					peerConnections[handshake.SessionID] = nil
+					delete(peerConnections, handshake.SessionID)
+					CandidatesMutex.Unlock()
+				} else if connectionState == pionWebRTC.PeerConnectionStateConnected {
+					CandidatesMutex.Lock()
 					atomic.AddInt64(&peerConnectionCount, 1)
-				} else if connectionState == pionWebRTC.ICEConnectionStateChecking {
-					// Iterate over the candidates and send them to the remote client
-					// Non blocking channel
-					for candidate := range CandidateArrays[sessionKey] {
-						log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Received candidate from channel: " + candidate)
-						if candidateErr := peerConnection.AddICECandidate(pionWebRTC.ICECandidateInit{Candidate: string(candidate)}); candidateErr != nil {
-							log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding candidate: " + candidateErr.Error())
-						}
-					}
-				} else if connectionState == pionWebRTC.ICEConnectionStateFailed {
+					CandidatesMutex.Unlock()
+				} else if connectionState == pionWebRTC.PeerConnectionStateFailed {
 					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICEConnectionStateFailed")
 				}
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection state changed to: " + connectionState.String())
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): Number of peers connected (" + strconv.FormatInt(peerConnectionCount, 10) + ")")
 			})
 
+			go func() {
+				// Iterate over the candidates and send them to the remote client
+				// Non blocking channe
+				for candidate := range CandidateArrays[sessionKey] {
+					CandidatesMutex.Lock()
+					log.Log.Info(">>>> webrtc.main.InitializeWebRTCConnection(): Received candidate from channel: " + candidate)
+					if candidateErr := peerConnection.AddICECandidate(pionWebRTC.ICECandidateInit{Candidate: string(candidate)}); candidateErr != nil {
+						log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding candidate: " + candidateErr.Error())
+					}
+					CandidatesMutex.Unlock()
+				}
+			}()
+
 			offer := w.CreateOffer(sd)
 			if err = peerConnection.SetRemoteDescription(offer); err != nil {
 				log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while setting remote description: " + err.Error())
@@ -220,11 +212,10 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				if candidate == nil {
 					return
 				}
+
 				//  Create a config map
 				valueMap := make(map[string]interface{})
 				candateJSON := candidate.ToJSON()
-				sdpmid := "0"
-				candateJSON.SDPMid = &sdpmid
 				candateBinary, err := json.Marshal(candateJSON)
 				if err == nil {
 					valueMap["candidate"] = string(candateBinary)
@@ -410,7 +401,7 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 				}
 				if start {
 					bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
-					sample := pionMedia.Sample{Data: pkt.Data, Duration: bufferDurationCasted}
+					sample := pionMedia.Sample{Data: pkt.Data, Duration: bufferDurationCasted, PacketTimestamp: uint32(pkt.Time)}
 					//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
 					if config.Capture.ForwardWebRTC == "true" {
 						// We will send the video to a remote peer
@@ -431,6 +422,7 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 					// We will transcode the audio
 					// TODO..
 					//d := fdkaac.NewAacDecoder()
+					continue
 				}
 
 				// Calculate the difference
@@ -439,7 +431,7 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 
 				// We will send the audio
 				bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
-				sample := pionMedia.Sample{Data: pkt.Data, Duration: bufferDurationCasted}
+				sample := pionMedia.Sample{Data: pkt.Data, Duration: bufferDurationCasted, PacketTimestamp: uint32(pkt.Time)}
 				//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
 				if err := audioTrack.WriteSample(sample); err != nil && err != io.ErrClosedPipe {
 					log.Log.Error("webrtc.main.WriteToTrack(): something went wrong while writing sample: " + err.Error())
@@ -447,11 +439,6 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 			}
 		}
 	}
-	for _, p := range peerConnections {
-		if p != nil {
-			p.Close()
-		}
-	}
 
 	peerConnectionCount = 0
 	log.Log.Info("webrtc.main.WriteToTrack(): stop writing to track.")