Merge pull request #158 from kerberos-io/fix/force-mqtt-tos-level-2

Enabled TOS 2 for MQTT to ensure higher quality

machinery/src/cloud/Cloud.go

@@ -6,7 +6,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
-	"image"
 	"io"
 	"os"
 	"strings"
@@ -690,28 +689,25 @@ 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{
@@ -728,8 +724,6 @@ 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 {
@@ -830,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/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.1"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"