do not read if not needed

Dockerfile

@@ -61,7 +61,7 @@ RUN /dist/agent/main version
 ###############################################
 # Build Bento4 -> we want fragmented mp4 files
 
-ENV BENTO4_VERSION 1.6.0-641
+ENV BENTO4_VERSION 1.6.0-639
 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. Bravo!
+This project exists thanks to all the people who contribute.
 
 <a href="https://github.com/kerberos-io/agent/graphs/contributors">
   <img src="https://contrib.rocks/image?repo=kerberos-io/agent" />

machinery/src/cloud/Cloud.go

@@ -6,6 +6,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"image"
 	"io"
 	"os"
 	"strings"
@@ -689,25 +690,28 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 			var pkt packets.Packet
 
 			for cursorError == nil {
-				pkt, cursorError = livestreamCursor.ReadPacket()
-				if len(pkt.Data) == 0 || !pkt.IsKeyFrame {
-					continue
-				}
+
 				now := time.Now().Unix()
 				select {
 				case <-communication.HandleLiveSD:
 					lastLivestreamRequest = now
 				default:
 				}
+
 				if now-lastLivestreamRequest > 3 {
 					continue
 				}
+
+				pkt, cursorError = livestreamCursor.ReadPacket()
+				if len(pkt.Data) == 0 || !pkt.IsKeyFrame {
+					continue
+				}
+
 				log.Log.Info("cloud.HandleLiveStreamSD(): Sending base64 encoded images to MQTT.")
 				img, err := rtspClient.DecodePacket(pkt)
 				if err == nil {
 					bytes, _ := utils.ImageToBytes(&img)
 					encoded := base64.StdEncoding.EncodeToString(bytes)
-
 					valueMap := make(map[string]interface{})
 					valueMap["image"] = encoded
 					message := models.Message{
@@ -724,6 +728,8 @@ func HandleLiveStreamSD(livestreamCursor *packets.QueueCursor, configuration *mo
 						log.Log.Info("cloud.HandleLiveStreamSD(): something went wrong while sending acknowledge config to hub: " + string(payload))
 					}
 				}
+				// Cleaning image
+				img = image.YCbCr{}
 			}
 
 		} else {
@@ -824,6 +830,7 @@ 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, 2, false, payload)
+											mqttClient.Publish("kerberos/hub/"+hubKey, 0, false, payload)
 										} else {
 											log.Log.Info("computervision.main.ProcessMotion(): failed to package MQTT message: " + err.Error())
 										}

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, 2, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 0, 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, 2, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 0, 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, 2, false, payload)
+				mqttClient.Publish("kerberos/hub/"+hubKey, 0, 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.3.5"
+const VERSION = "3.2.6"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
 

machinery/src/webrtc/main.go

@@ -64,6 +64,7 @@ func CreateWebRTC(name string, stunServers []string, turnServers []string, turnS
 		TurnServersUsername:   turnServersUsername,
 		TurnServersCredential: turnServersCredential,
 		Timer:                 time.NewTimer(time.Second * 10),
+		PacketsCount:          make(chan int),
 	}
 }
 
@@ -166,37 +167,35 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				log.Log.Error("webrtc.main.InitializeWebRTCConnection(): something went wrong while adding audio track: " + err.Error())
 			}
 
-			peerConnection.OnConnectionStateChange(func(connectionState pionWebRTC.PeerConnectionState) {
-				if connectionState == pionWebRTC.PeerConnectionStateDisconnected || connectionState == pionWebRTC.PeerConnectionStateClosed {
+			peerConnection.OnICEConnectionStateChange(func(connectionState pionWebRTC.ICEConnectionState) {
+				if connectionState == pionWebRTC.ICEConnectionStateDisconnected {
+					atomic.AddInt64(&peerConnectionCount, -1)
+
 					// Set lock
 					CandidatesMutex.Lock()
-					atomic.AddInt64(&peerConnectionCount, -1)
 					peerConnections[handshake.SessionID] = nil
 					_, 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())
 					}
-					CandidatesMutex.Unlock()
-				} else if connectionState == pionWebRTC.PeerConnectionStateConnected {
-					CandidatesMutex.Lock()
+				} else if connectionState == pionWebRTC.ICEConnectionStateConnected {
 					atomic.AddInt64(&peerConnectionCount, 1)
-					CandidatesMutex.Unlock()
-				} else if connectionState == pionWebRTC.PeerConnectionStateConnecting {
+				} else if connectionState == pionWebRTC.ICEConnectionStateChecking {
 					// Iterate over the candidates and send them to the remote client
-					// Non blocking channe
+					// Non blocking channel
 					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.PeerConnectionStateFailed {
+				} else if connectionState == pionWebRTC.ICEConnectionStateFailed {
 					log.Log.Info("webrtc.main.InitializeWebRTCConnection(): ICEConnectionStateFailed")
 				}
 				log.Log.Info("webrtc.main.InitializeWebRTCConnection(): connection state changed to: " + connectionState.String())
@@ -218,10 +217,9 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 			// 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) {
-				if candidate == nil || peerConnection.ICEConnectionState() == pionWebRTC.ICEConnectionStateConnected {
+				if candidate == nil {
 					return
 				}
-
 				//  Create a config map
 				valueMap := make(map[string]interface{})
 				candateJSON := candidate.ToJSON()
@@ -273,7 +271,6 @@ func InitializeWebRTCConnection(configuration *models.Configuration, communicati
 				}
 				payload, err := models.PackageMQTTMessage(configuration, message)
 				if err == nil {
-					time.Sleep(1000 * time.Millisecond)
 					token := mqttClient.Publish("kerberos/hub/"+hubKey, 2, false, payload)
 					token.Wait()
 				} else {
@@ -344,8 +341,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
@@ -404,17 +401,16 @@ 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, Duration: bufferDurationCasted, PacketTimestamp: uint32(pkt.Time)}
-					//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
+					//bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
+					sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
 					if config.Capture.ForwardWebRTC == "true" {
 						// We will send the video to a remote peer
 						// TODO..
@@ -434,17 +430,15 @@ 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, Duration: bufferDurationCasted, PacketTimestamp: uint32(pkt.Time)}
-				//sample = pionMedia.Sample{Data: pkt.Data, Duration: time.Second}
+				//bufferDurationCasted := time.Duration(bufferDuration) * time.Millisecond
+				sample := pionMedia.Sample{Data: pkt.Data, PacketTimestamp: uint32(pkt.Time)}
 				if err := audioTrack.WriteSample(sample); err != nil && err != io.ErrClosedPipe {
 					log.Log.Error("webrtc.main.WriteToTrack(): something went wrong while writing sample: " + err.Error())
 				}