unsubscribe from mqtt subscriptions

machinery/src/components/Config.go

@@ -56,6 +56,9 @@ func GetSnapshot() string {
 		content, _ := ioutil.ReadAll(reader)
 		// Encode as base64.
 		snapshot = base64.StdEncoding.EncodeToString(content)
+
+		// Close reader
+		reader = nil
 	}
 	return snapshot
 }
@@ -146,6 +149,9 @@ func OpenConfig(configuration *models.Configuration) {
 		conjungo.Merge(&s3, configuration.CustomConfig.S3, opts)
 		configuration.Config.S3 = &s3
 
+		// Cleanup
+		opts = nil
+
 	} else if os.Getenv("DEPLOYMENT") == "" || os.Getenv("DEPLOYMENT") == "agent" {
 
 		// Local deployment means we do a stand-alone installation

machinery/src/components/Kerberos.go

@@ -222,7 +222,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 		close(communication.HandleONVIF)
 		close(communication.HandleLiveHDHandshake)
 		close(communication.HandleMotion)
-		routers.DisconnectMQTT(mqttClient)
+		routers.DisconnectMQTT(mqttClient, &configuration.Config)
 
 		// Wait a few seconds to stop the decoder.
 		time.Sleep(time.Second * 3)

machinery/src/computervision/main.go

@@ -130,26 +130,7 @@ func ProcessMotion(motionCursor *pubsub.QueueCursor, configuration *models.Confi
 
 				// Store snapshots (jpg) for hull.
 				if config.Capture.Snapshots != "false" {
-					files, err := ioutil.ReadDir("./data/snapshots")
-					if err == nil {
-						rgbImage, err := GetRawImage(frame, pkt, decoder, decoderMutex)
-						if err == nil {
-							sort.Slice(files, func(i, j int) bool {
-								return files[i].ModTime().Before(files[j].ModTime())
-							})
-							if len(files) > 3 {
-								os.Remove("./data/snapshots/" + files[0].Name())
-							}
-
-							// Save image
-							t := strconv.FormatInt(time.Now().Unix(), 10)
-							f, err := os.Create("./data/snapshots/" + t + ".jpg")
-							if err == nil {
-								jpeg.Encode(f, &rgbImage.Image, &jpeg.Options{Quality: 15})
-								f.Close()
-							}
-						}
-					}
+					StoreSnapshot(frame, pkt, decoder, decoderMutex)
 				}
 
 				// Check if within time interval
@@ -255,3 +236,26 @@ func AbsDiffBitwiseAndThreshold(img1 *image.Gray, img2 *image.Gray, img3 *image.
 	}
 	return changes
 }
+
+func StoreSnapshot(frame *ffmpeg.VideoFrame, pkt av.Packet, decoder *ffmpeg.VideoDecoder, decoderMutex *sync.Mutex) {
+	files, err := ioutil.ReadDir("./data/snapshots")
+	if err == nil {
+		rgbImage, err := GetRawImage(frame, pkt, decoder, decoderMutex)
+		if err == nil {
+			sort.Slice(files, func(i, j int) bool {
+				return files[i].ModTime().Before(files[j].ModTime())
+			})
+			if len(files) > 3 {
+				os.Remove("./data/snapshots/" + files[0].Name())
+			}
+
+			// Save image
+			t := strconv.FormatInt(time.Now().Unix(), 10)
+			f, err := os.Create("./data/snapshots/" + t + ".jpg")
+			if err == nil {
+				jpeg.Encode(f, &rgbImage.Image, &jpeg.Options{Quality: 15})
+				f.Close()
+			}
+		}
+	}
+}

machinery/src/routers/mqtt/main.go

@@ -193,8 +193,16 @@ func MQTTListenerHandleONVIF(mqttClient mqtt.Client, hubKey string, configuratio
 	})
 }
 
-func DisconnectMQTT(mqttClient mqtt.Client) {
+func DisconnectMQTT(mqttClient mqtt.Client, config *models.Config) {
 	if mqttClient != nil {
+		// Cleanup all subscriptions.
+		mqttClient.Unsubscribe("kerberos/" + config.HubKey + "/device/" + config.Key + "/request-live")
+		mqttClient.Unsubscribe(config.Key + "/register")
+		mqttClient.Unsubscribe("kerberos/webrtc/keepalivehub/" + config.Key)
+		mqttClient.Unsubscribe("kerberos/webrtc/peers/" + config.Key)
+		mqttClient.Unsubscribe("candidate/cloud")
+		mqttClient.Unsubscribe("kerberos/onvif/" + config.Key)
 		mqttClient.Disconnect(1000)
+		mqttClient = nil
 	}
 }