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

Enabled TOS 2 for MQTT to ensure higher quality

.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

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.1"
 
 const letterBytes = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"