webrtc: disable relay allow other

machinery/src/cloud/Cloud.go

@@ -554,10 +554,12 @@ func HandleLiveStreamHD(livestreamCursor *pubsub.QueueCursor, configuration *mod
 				for handshake := range communication.HandleLiveHDHandshake {
 					log.Log.Info("HandleLiveStreamHD: setting up a peer connection.")
 					key := config.Key + "/" + handshake.SessionID
+					webrtc.CandidatesMutex.Lock()
 					_, ok := webrtc.CandidateArrays[key]
 					if !ok {
 						webrtc.CandidateArrays[key] = make(chan string)
 					}
+					webrtc.CandidatesMutex.Unlock()
 					webrtc.InitializeWebRTCConnection(configuration, communication, mqttClient, videoTrack, audioTrack, handshake, webrtc.CandidateArrays[key])
 
 				}

machinery/src/routers/mqtt/main.go

@@ -457,7 +457,12 @@ func HandleReceiveHDCandidates(mqttClient mqtt.Client, hubKey string, payload mo
 
 	if receiveHDCandidatesPayload.Timestamp != 0 {
 		if communication.CameraConnected {
-			channel := webrtc.CandidateArrays[receiveHDCandidatesPayload.SessionID]
+			key := configuration.Config.Key + "/" + receiveHDCandidatesPayload.SessionID
+			channel := webrtc.CandidateArrays[key]
+			if channel == nil {
+				channel = make(chan string)
+				webrtc.CandidateArrays[key] = channel
+			}
 			log.Log.Info("HandleReceiveHDCandidates: " + receiveHDCandidatesPayload.Candidate)
 			channel <- receiveHDCandidatesPayload.Candidate
 		} else {

machinery/src/webrtc/main.go

@@ -125,6 +125,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 						Credential: w.TurnServersCredential,
 					},
 				},
+				//ICETransportPolicy: pionWebRTC.ICETransportPolicyRelay,
 			},
 		)
 
@@ -179,6 +180,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				panic(err)
 			}
 
+			// When an ICE candidate is available send to the other Pion instance
+			// the other Pion instance will add this candidate by calling AddICECandidate
+			var candidatesMux sync.Mutex
 			// When an ICE candidate is available send to the other peer using the signaling server (MQTT).
 			// The other peer will add this candidate by calling AddICECandidate
 			peerConnection.OnICECandidate(func(candidate *pionWebRTC.ICECandidate) {
@@ -186,6 +190,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 					return
 				}
 
+				candidatesMux.Lock()
+				defer candidatesMux.Unlock()
+
 				//  Create a config map
 				valueMap := make(map[string]interface{})
 				candateJSON := candidate.ToJSON()
@@ -209,7 +216,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				payload, err := models.PackageMQTTMessage(configuration, message)
 				if err == nil {
 					log.Log.Info("InitializeWebRTCConnection:" + string(candateBinary))
-					token := mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+					token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 					token.Wait()
 				} else {
 					log.Log.Info("HandleRequestConfig: something went wrong while sending acknowledge config to hub: " + string(payload))
@@ -235,7 +242,7 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				}
 				payload, err := models.PackageMQTTMessage(configuration, message)
 				if err == nil {
-					token := mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+					token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 					token.Wait()
 				} else {
 					log.Log.Info("HandleRequestConfig: something went wrong while sending acknowledge config to hub: " + string(payload))