do allow more candidates even after connection state, remove sdpmid + remove sleep and unused code.

Correct peerconnection states + proper cleanup peerconnection

.github/workflows/docker-dev.yml

@@ -14,8 +14,8 @@ jobs:
       - name: Login to DockerHub
         uses: docker/login-action@v2
         with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Checkout
         uses: actions/checkout@v3
       - name: Set up QEMU
@@ -40,8 +40,8 @@ jobs:
       - name: Login to DockerHub
         uses: docker/login-action@v2
         with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Checkout
         uses: actions/checkout@v3
       - name: Set up QEMU

.github/workflows/docker-nightly.yml

@@ -15,8 +15,8 @@ jobs:
       - name: Login to DockerHub
         uses: docker/login-action@v2
         with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Checkout
         run: git clone https://github.com/kerberos-io/agent && cd agent
       - name: Set up QEMU
@@ -38,8 +38,8 @@ jobs:
       - name: Login to DockerHub
         uses: docker/login-action@v2
         with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Checkout
         run: git clone https://github.com/kerberos-io/agent && cd agent
       - name: Set up QEMU

.github/workflows/docker.yml

@@ -19,8 +19,8 @@ jobs:
       - name: Login to DockerHub
         uses: docker/login-action@v2
         with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Checkout
         uses: actions/checkout@v3
       - uses: benjlevesque/short-sha@v2.1
@@ -77,8 +77,8 @@ jobs:
       - name: Login to DockerHub
         uses: docker/login-action@v2
         with:
-          username: ${{ secrets.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
       - name: Checkout
         uses: actions/checkout@v3
       - uses: benjlevesque/short-sha@v2.1

.github/workflows/pr-description.yaml

@@ -0,0 +1,19 @@
+name: Autofill PR description
+
+on: pull_request
+
+jobs:
+  openai-pr-description:
+    runs-on: ubuntu-22.04
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Autofill PR description if empty using OpenAI
+        uses: cedricve/azureopenai-pr-description@master
+        with:
+          github_token: ${{ secrets.TOKEN }}
+          openai_api_key: ${{ secrets.OPENAI_API_KEY }}
+          azure_openai_api_key: ${{ secrets.AZURE_OPENAI_API_KEY }}
+          azure_openai_endpoint: ${{ secrets.AZURE_OPENAI_ENDPOINT }}
+          azure_openai_version: ${{ secrets.AZURE_OPENAI_VERSION }}
+          overwrite_description: true

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 && \

README.md

@@ -413,7 +413,7 @@ Read more about this [at the FAQ](#faq) below.
 
 ## Contributors
 
-This project exists thanks to all the people who contribute.
+This project exists thanks to all the people who contribute. Bravo!
 
 <a href="https://github.com/kerberos-io/agent/graphs/contributors">
   <img src="https://contrib.rocks/image?repo=kerberos-io/agent" />

machinery/go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/gorilla/websocket v1.5.3
 	github.com/kellydunn/golang-geo v0.7.0
 	github.com/kerberos-io/joy4 v1.0.64
-	github.com/kerberos-io/onvif v0.0.16
+	github.com/kerberos-io/onvif v1.0.0
 	github.com/minio/minio-go/v6 v6.0.57
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
 	github.com/pion/rtp v1.8.10

machinery/go.sum

@@ -298,8 +298,8 @@ github.com/kellydunn/golang-geo v0.7.0 h1:A5j0/BvNgGwY6Yb6inXQxzYwlPHc6WVZR+Mrar
 github.com/kellydunn/golang-geo v0.7.0/go.mod h1:YYlQPJ+DPEzrHx8kT3oPHC/NjyvCCXE+IuKGKdrjrcU=
 github.com/kerberos-io/joy4 v1.0.64 h1:gTUSotHSOhp9mNqEecgq88tQHvpj7TjmrvPUsPm0idg=
 github.com/kerberos-io/joy4 v1.0.64/go.mod h1:nZp4AjvKvTOXRrmDyAIOw+Da+JA5OcSo/JundGfOlFU=
-github.com/kerberos-io/onvif v0.0.16 h1:fSiyc5ja6Q6ux6Jscvv4TVDPeK8O9x8WT1fG7YsbC8Q=
-github.com/kerberos-io/onvif v0.0.16/go.mod h1:P1kUcCfeotJSlL1jwGseH6NSnCwWiuJLl3gAzafnLbA=
+github.com/kerberos-io/onvif v1.0.0 h1:pLJrK6skPkK+5Bj4XfqHUkQ2I+p5pwELnp+kQTJWXiQ=
+github.com/kerberos-io/onvif v1.0.0/go.mod h1:P1kUcCfeotJSlL1jwGseH6NSnCwWiuJLl3gAzafnLbA=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.17.1 h1:NE3C767s2ak2bweCZo3+rdP4U/HoyVXLv/X9f2gPS5g=

machinery/src/capture/gortsplib.go

@@ -83,6 +83,25 @@ type Golibrtsp struct {
 	Streams []packets.Stream
 }
 
+// Init function
+var H264FrameDecoder *Decoder
+var H265FrameDecoder *Decoder
+
+func init() {
+	var err error
+	// setup H264 -> raw frames decoder
+	H264FrameDecoder, err = newDecoder("H264")
+	if err != nil {
+		log.Log.Error("capture.golibrtsp.init(): " + err.Error())
+	}
+
+	// setup H265 -> raw frames decoder
+	H265FrameDecoder, err = newDecoder("H265")
+	if err != nil {
+		log.Log.Error("capture.golibrtsp.init(): " + err.Error())
+	}
+}
+
 // Connect to the RTSP server.
 func (g *Golibrtsp) Connect(ctx context.Context) (err error) {
 
@@ -171,13 +190,7 @@ func (g *Golibrtsp) Connect(ctx context.Context) (err error) {
 				log.Log.Error("capture.golibrtsp.Connect(H264): " + err.Error())
 			}
 			g.VideoH264Decoder = rtpDec
-
-			// setup H264 -> raw frames decoder
-			frameDec, err := newDecoder("H264")
-			if err != nil {
-				log.Log.Error("capture.golibrtsp.Connect(H264): " + err.Error())
-			}
-			g.VideoH264FrameDecoder = frameDec
+			g.VideoH264FrameDecoder = H264FrameDecoder
 		}
 	}
 
@@ -227,12 +240,7 @@ func (g *Golibrtsp) Connect(ctx context.Context) (err error) {
 			}
 			g.VideoH265Decoder = rtpDec
 
-			// setup H265 -> raw frames decoder
-			frameDec, err := newDecoder("H265")
-			if err != nil {
-				log.Log.Error("capture.golibrtsp.Connect(H265): " + err.Error())
-			}
-			g.VideoH265FrameDecoder = frameDec
+			g.VideoH265FrameDecoder = H265FrameDecoder
 		}
 	}
 
@@ -882,12 +890,15 @@ func (g *Golibrtsp) GetAudioStreams() ([]packets.Stream, error) {
 func (g *Golibrtsp) Close() error {
 	// Close the demuxer.
 	g.Client.Close()
-	if g.VideoH264Decoder != nil {
-		g.VideoH264FrameDecoder.Close()
-	}
-	if g.VideoH265FrameDecoder != nil {
-		g.VideoH265FrameDecoder.Close()
-	}
+
+	// We will have created the decoders globally, so we don't need to close them here.
+
+	//if g.VideoH264Decoder != nil {
+	//	g.VideoH264FrameDecoder.Close()
+	//}
+	//if g.VideoH265FrameDecoder != nil {
+	//	g.VideoH265FrameDecoder.Close()
+	//}
 	return nil
 }
 

machinery/src/cloud/Cloud.go

@@ -824,7 +824,6 @@ func HandleRealtimeProcessing(processingCursor *packets.QueueCursor, configurati
 					}
 					payload, err := models.PackageMQTTMessage(configuration, message)
 					if err == nil {
-
 						mqttClient.Publish(realtimeProcessingTopic, 0, false, payload)
 					} else {
 						log.Log.Info("cloud.RealtimeProcessing(): something went wrong while sending acknowledge config to hub: " + string(payload))

machinery/src/computervision/main.go

@@ -150,7 +150,7 @@ func ProcessMotion(motionCursor *packets.QueueCursor, configuration *models.Conf
 										}
 										payload, err := models.PackageMQTTMessage(configuration, message)
 										if err == nil {
-											mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+											mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 										} else {
 											log.Log.Info("computervision.main.ProcessMotion(): failed to package MQTT message: " + err.Error())
 										}

machinery/src/onvif/main.go

@@ -222,10 +222,10 @@ func ConnectToOnvifDevice(cameraConfiguration *models.IPCamera) (*onvif.Device,
 		var b []byte
 		if resp != nil {
 			b, err = io.ReadAll(resp.Body)
+			resp.Body.Close() // Ensure the response body is closed
 			if err != nil {
 				log.Log.Error("onvif.ConnectToOnvifDevice(): " + err.Error())
 			}
-			resp.Body.Close()
 		}
 		stringBody := string(b)
 		decodedXML, et, err := getXMLNode(stringBody, "GetCapabilitiesResponse")
@@ -252,10 +252,10 @@ func GetTokenFromProfile(device *onvif.Device, profileId int) (xsdonvif.Referenc
 	// Get Profiles
 	resp, err := device.CallMethod(media.GetProfiles{})
 	if err == nil {
-		defer resp.Body.Close()
 		b, err := io.ReadAll(resp.Body)
 		if err == nil {
 			stringBody := string(b)
+			resp.Body.Close() // Ensure the response body is closed
 			decodedXML, et, err := getXMLNode(stringBody, "GetProfilesResponse")
 			if err != nil {
 				log.Log.Debug("onvif.GetTokenFromProfile(): " + err.Error())
@@ -288,21 +288,19 @@ func GetPTZConfigurationsFromDevice(device *onvif.Device) (ptz.GetConfigurations
 	var b []byte
 	if resp != nil {
 		b, err = io.ReadAll(resp.Body)
-		resp.Body.Close()
+		resp.Body.Close() // Ensure the response body is closed
 	}
 
 	if err == nil {
-		if err == nil {
-			stringBody := string(b)
-			decodedXML, et, err := getXMLNode(stringBody, "GetConfigurationsResponse")
-			if err != nil {
+		stringBody := string(b)
+		decodedXML, et, err := getXMLNode(stringBody, "GetConfigurationsResponse")
+		if err != nil {
+			log.Log.Debug("onvif.GetPTZConfigurationsFromDevice(): " + err.Error())
+			return configurations, err
+		} else {
+			if err := decodedXML.DecodeElement(&configurations, et); err != nil {
 				log.Log.Debug("onvif.GetPTZConfigurationsFromDevice(): " + err.Error())
 				return configurations, err
-			} else {
-				if err := decodedXML.DecodeElement(&configurations, et); err != nil {
-					log.Log.Debug("onvif.GetPTZConfigurationsFromDevice(): " + err.Error())
-					return configurations, err
-				}
 			}
 		}
 	}
@@ -360,7 +358,7 @@ func GetPosition(device *onvif.Device, token xsdonvif.ReferenceToken) (xsdonvif.
 	var b []byte
 	if resp != nil {
 		b, err = io.ReadAll(resp.Body)
-		resp.Body.Close()
+		resp.Body.Close() // Ensure the response body is closed
 	}
 
 	if err == nil {
@@ -794,7 +792,7 @@ func GetPresetsFromDevice(device *onvif.Device) ([]models.OnvifActionPreset, err
 		var b []byte
 		if resp != nil {
 			b, err = io.ReadAll(resp.Body)
-			resp.Body.Close()
+			resp.Body.Close() // Ensure the response body is closed
 		}
 		if err == nil {
 			stringBody := string(b)
@@ -845,7 +843,7 @@ func GoToPresetFromDevice(device *onvif.Device, presetName string) error {
 		var b []byte
 		if resp != nil {
 			b, err = io.ReadAll(resp.Body)
-			resp.Body.Close()
+			resp.Body.Close() // Ensure the response body is closed
 		}
 		if err == nil {
 			stringBody := string(b)
@@ -1004,7 +1002,7 @@ func CreatePullPointSubscription(dev *onvif.Device) (string, error) {
 	var b2 []byte
 	if resp != nil {
 		b2, err = io.ReadAll(resp.Body)
-		resp.Body.Close()
+		resp.Body.Close() // Ensure the response body is closed
 		if err == nil {
 			stringBody := string(b2)
 			decodedXML, et, err := getXMLNode(stringBody, "CreatePullPointSubscriptionResponse")
@@ -1023,19 +1021,25 @@ func CreatePullPointSubscription(dev *onvif.Device) (string, error) {
 }
 
 func UnsubscribePullPoint(dev *onvif.Device, pullPointAddress string) error {
+
 	// Unsubscribe from the device
 	unsubscribe := event.Unsubscribe{}
 	requestBody, err := xml.Marshal(unsubscribe)
 	if err != nil {
 		log.Log.Error("onvif.main.UnsubscribePullPoint(): " + err.Error())
 	}
+
 	res, err := dev.SendSoap(pullPointAddress, string(requestBody))
 	if err != nil {
 		log.Log.Error("onvif.main.UnsubscribePullPoint(): " + err.Error())
 	}
 	if res != nil {
-		_, err := io.ReadAll(res.Body)
-		res.Body.Close()
+		b, err := io.ReadAll(res.Body)
+		res.Body.Close() // Ensure the response body is closed
+		if err == nil {
+			stringBody := string(b)
+			log.Log.Debug("onvif.main.UnsubscribePullPoint(): " + stringBody)
+		}
 		if err != nil {
 			log.Log.Error("onvif.main.UnsubscribePullPoint(): " + err.Error())
 		}
@@ -1099,7 +1103,7 @@ func GetEventMessages(dev *onvif.Device, pullPointAddress string) ([]ONVIFEvents
 			var pullMessagesResponse event.PullMessagesResponse
 			if res != nil {
 				bs, err := io.ReadAll(res.Body)
-				res.Body.Close()
+				res.Body.Close() // Ensure the response body is closed
 				if err == nil {
 					stringBody := string(bs)
 					decodedXML, et, err := getXMLNode(stringBody, "PullMessagesResponse")
@@ -1208,7 +1212,7 @@ func GetDigitalInputs(dev *onvif.Device) (device.GetDigitalInputsResponse, error
 	resp, err := dev.CallMethod(deviceio.GetDigitalInputs{})
 	if resp != nil {
 		b, err = io.ReadAll(resp.Body)
-		resp.Body.Close()
+		resp.Body.Close() // Ensure the response body is closed
 	}
 
 	if err == nil {
@@ -1240,21 +1244,19 @@ func GetRelayOutputs(dev *onvif.Device) (device.GetRelayOutputsResponse, error)
 	var b []byte
 	if resp != nil {
 		b, err = io.ReadAll(resp.Body)
-		resp.Body.Close()
+		resp.Body.Close() // Ensure the response body is closed
 	}
 
 	if err == nil {
-		if err == nil {
-			stringBody := string(b)
-			decodedXML, et, err := getXMLNode(stringBody, "GetRelayOutputsResponse")
-			if err != nil {
-				log.Log.Error("onvif.main.GetRelayOutputs(): " + err.Error())
+		stringBody := string(b)
+		decodedXML, et, err := getXMLNode(stringBody, "GetRelayOutputsResponse")
+		if err != nil {
+			log.Log.Error("onvif.main.GetRelayOutputs(): " + err.Error())
+			return relayoutputs, err
+		} else {
+			if err := decodedXML.DecodeElement(&relayoutputs, et); err != nil {
+				log.Log.Debug("onvif.main.GetRelayOutputs(): " + err.Error())
 				return relayoutputs, err
-			} else {
-				if err := decodedXML.DecodeElement(&relayoutputs, et); err != nil {
-					log.Log.Debug("onvif.main.GetRelayOutputs(): " + err.Error())
-					return relayoutputs, err
-				}
 			}
 		}
 	}
@@ -1283,7 +1285,7 @@ func TriggerRelayOutput(dev *onvif.Device, output string) (err error) {
 			var b []byte
 			if errResp != nil {
 				b, err = io.ReadAll(resp.Body)
-				resp.Body.Close()
+				resp.Body.Close() // Ensure the response body is closed
 			}
 			stringBody := string(b)
 			if err == nil && resp.StatusCode == 200 {

machinery/src/routers/http/Server.go

@@ -47,7 +47,7 @@ func StartServer(configDirectory string, configuration *models.Configuration, co
 	// Initialize REST API
 	r := gin.Default()
 
-	// Profileerggerg
+	// Profiler
 	pprof.Register(r)
 
 	// Setup CORS

machinery/src/routers/mqtt/main.go

@@ -341,7 +341,7 @@ func HandleGetPTZPosition(mqttClient mqtt.Client, hubKey string, payload models.
 			}
 			payload, err := models.PackageMQTTMessage(configuration, message)
 			if err == nil {
-				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 			} else {
 				log.Log.Info("routers.mqtt.main.HandlePTZPosition(): something went wrong while sending position to hub: " + string(payload))
 			}
@@ -412,7 +412,7 @@ func HandleRequestConfig(mqttClient mqtt.Client, hubKey string, payload models.P
 			}
 			payload, err := models.PackageMQTTMessage(configuration, message)
 			if err == nil {
-				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 			} else {
 				log.Log.Info("routers.mqtt.main.HandleRequestConfig(): something went wrong while sending config to hub: " + string(payload))
 			}
@@ -451,7 +451,7 @@ func HandleUpdateConfig(mqttClient mqtt.Client, hubKey string, payload models.Pa
 			}
 			payload, err := models.PackageMQTTMessage(configuration, message)
 			if err == nil {
-				mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 			} else {
 				log.Log.Info("routers.mqtt.main.HandleUpdateConfig(): something went wrong while sending acknowledge config to hub: " + string(payload))
 			}

machinery/src/utils/main.go

@@ -23,7 +23,7 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/models"
 )
 
-const VERSION = "3.2.6"
+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),
 	}
 }
 
@@ -167,35 +147,37 @@ 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)
 					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
+					CandidatesMutex.Unlock()
+				} else if connectionState == pionWebRTC.PeerConnectionStateConnected {
+					CandidatesMutex.Lock()
 					atomic.AddInt64(&peerConnectionCount, 1)
-				} else if connectionState == pionWebRTC.ICEConnectionStateChecking {
+					CandidatesMutex.Unlock()
+				} else if connectionState == pionWebRTC.PeerConnectionStateConnecting {
 					// Iterate over the candidates and send them to the remote client
-					// Non blocking channel
+					// 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()
 					}
-				} else if connectionState == pionWebRTC.ICEConnectionStateFailed {
+				} else if connectionState == pionWebRTC.PeerConnectionStateFailed {
 					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICEConnectionStateFailed")
 				}
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection state changed to: " + connectionState.String())
@@ -220,11 +202,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)
@@ -341,8 +322,8 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 
 		var cursorError error
 		var pkt packets.Packet
-		//var previousTimeVideo int64
-		//var previousTimeAudio int64
+		var previousTimeVideo int64
+		var previousTimeAudio int64
 
 		start := false
 		receivedKeyFrame := false
@@ -401,16 +382,17 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 			if pkt.IsVideo {
 
 				// Calculate the difference
-				//bufferDuration := pkt.Time - previousTimeVideo
-				//previousTimeVideo = pkt.Time
+				bufferDuration := pkt.Time - previousTimeVideo
+				previousTimeVideo = pkt.Time
 
 				// Start at the first keyframe
 				if pkt.IsKeyFrame {
 					start = true
 				}
 				if start {
-					//bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
-					sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
+					bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
+					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
 						// TODO..
@@ -430,15 +412,17 @@ func WriteToTrack(livestreamCursor *packets.QueueCursor, configuration *models.C
 					// We will transcode the audio
 					// TODO..
 					//d := fdkaac.NewAacDecoder()
+					continue
 				}
 
 				// Calculate the difference
-				//bufferDuration := pkt.Time - previousTimeAudio
-				//previousTimeAudio = pkt.Time
+				bufferDuration := pkt.Time - previousTimeAudio
+				previousTimeAudio = pkt.Time
 
 				// We will send the audio
-				//bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
-				sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
+				bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
+				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())
 				}