introduce camera connected variable + allow MQTT stream to be connected even when no camera attached

machinery/src/components/Config.go

@@ -447,9 +447,11 @@ func SaveConfig(config models.Config, configuration *models.Configuration, commu
 			return err
 		}
 
-		select {
-		case communication.HandleBootstrap <- "restart":
-		default:
+		if communication.CameraConnected {
+			select {
+			case communication.HandleBootstrap <- "restart":
+			default:
+			}
 		}
 
 		communication.IsConfiguring.UnSet()

machinery/src/components/Kerberos.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/kerberos-io/joy4/cgo/ffmpeg"
 
 	"github.com/kerberos-io/agent/machinery/src/capture"
@@ -54,6 +55,7 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun
 	communication.HandleLiveSD = make(chan int64, 1)
 	communication.HandleLiveHDKeepalive = make(chan string, 1)
 	communication.HandleLiveHDPeers = make(chan string, 1)
+	communication.HandleONVIF = make(chan models.OnvifAction, 1)
 	communication.IsConfiguring = abool.New()
 
 	// Before starting the agent, we have a control goroutine, that might
@@ -68,23 +70,36 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun
 	// Handle heartbeats
 	go cloud.HandleHeartBeat(configuration, communication, uptimeStart)
 
+	// We'll create a MQTT handler, which will be used to communicate with Kerberos Hub.
+	// Configure a MQTT client which helps for a bi-directional communication
+	mqttClient := routers.ConfigureMQTT(configuration, communication)
+
 	// Run the agent and fire up all the other
 	// goroutines which do image capture, motion detection, onvif, etc.
-
 	for {
+
 		// This will blocking until receiving a signal to be restarted, reconfigured, stopped, etc.
-		status := RunAgent(configuration, communication, uptimeStart, cameraSettings, decoder, subDecoder)
+		status := RunAgent(configuration, communication, mqttClient, uptimeStart, cameraSettings, decoder, subDecoder)
 
 		if status == "stop" {
 			break
 		}
 
+		// We will reconfigure or restart the agent, we will mark the agent as not connected.
+		communication.CameraConnected = false
+
 		// We will re open the configuration, might have changed :O!
 		OpenConfig(configuration)
 
 		// We will override the configuration with the environment variables
 		OverrideWithEnvironmentVariables(configuration)
 
+		// Reset the MQTT client, might have provided new information, so we need to reconnect.
+		if routers.HasMQTTClientModified(configuration) {
+			routers.DisconnectMQTT(mqttClient, &configuration.Config)
+			mqttClient = routers.ConfigureMQTT(configuration, communication)
+		}
+
 		// We will create a new cancelable context, which will be used to cancel and restart.
 		// This is used to restart the agent when the configuration is updated.
 		ctx, cancel := context.WithCancel(context.Background())
@@ -94,7 +109,7 @@ func Bootstrap(configuration *models.Configuration, communication *models.Commun
 	log.Log.Debug("Bootstrap: finished")
 }
 
-func RunAgent(configuration *models.Configuration, communication *models.Communication, uptimeStart time.Time, cameraSettings *models.Camera, decoder *ffmpeg.VideoDecoder, subDecoder *ffmpeg.VideoDecoder) string {
+func RunAgent(configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, uptimeStart time.Time, cameraSettings *models.Camera, decoder *ffmpeg.VideoDecoder, subDecoder *ffmpeg.VideoDecoder) string {
 
 	log.Log.Debug("RunAgent: bootstrapping agent")
 	config := configuration.Config
@@ -129,7 +144,7 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 
 	if err == nil {
 
-		log.Log.Info("RunAgent: opened RTSP stream" + rtspUrl)
+		log.Log.Info("RunAgent: opened RTSP stream: " + rtspUrl)
 
 		// We might have a secondary rtsp url, so we might need to use that.
 		var subInfile av.DemuxCloser
@@ -209,9 +224,8 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 			subQueue.WriteHeader(subStreams)
 		}
 
-		// Configure a MQTT client which helps for a bi-directional communication
-		communication.HandleONVIF = make(chan models.OnvifAction, 1)
-		mqttClient := routers.ConfigureMQTT(configuration, communication)
+		// If we reach this point, we have a working RTSP connection.
+		communication.CameraConnected = true
 
 		// Handle the camera stream
 		go capture.HandleStream(infile, queue, communication)
@@ -291,16 +305,9 @@ func RunAgent(configuration *models.Configuration, communication *models.Communi
 			subQueue = nil
 			communication.SubQueue = nil
 		}
-		close(communication.HandleONVIF)
-		communication.HandleONVIF = nil
-		close(communication.HandleLiveHDHandshake)
-		communication.HandleLiveHDHandshake = nil
 		close(communication.HandleMotion)
 		communication.HandleMotion = nil
 
-		// Disconnect MQTT
-		routers.DisconnectMQTT(mqttClient, &configuration.Config)
-
 		// Wait a few seconds to stop the decoder.
 		time.Sleep(time.Second * 3)
 
@@ -329,29 +336,32 @@ func ControlAgent(communication *models.Communication) {
 		var previousPacket int64 = 0
 		var occurence = 0
 		for {
-			packetsR := packageCounter.Load().(int64)
-			if packetsR == previousPacket {
-				// If we are already reconfiguring,
-				// we dont need to check if the stream is blocking.
-				if !communication.IsConfiguring.IsSet() {
-					occurence = occurence + 1
+
+			// If camera is connected, we'll check if we are still receiving packets.
+			if communication.CameraConnected {
+				packetsR := packageCounter.Load().(int64)
+				if packetsR == previousPacket {
+					// If we are already reconfiguring,
+					// we dont need to check if the stream is blocking.
+					if !communication.IsConfiguring.IsSet() {
+						occurence = occurence + 1
+					}
+				} else {
+					occurence = 0
 				}
-			} else {
 
-				occurence = 0
+				log.Log.Info("ControlAgent: Number of packets read " + strconv.FormatInt(packetsR, 10))
+
+				// After 15 seconds without activity this is thrown..
+				if occurence == 3 {
+					log.Log.Info("Main: Restarting machinery.")
+					communication.HandleBootstrap <- "restart"
+					time.Sleep(2 * time.Second)
+					occurence = 0
+				}
+				previousPacket = packageCounter.Load().(int64)
 			}
 
-			log.Log.Info("ControlAgent: Number of packets read " + strconv.FormatInt(packetsR, 10))
-
-			// After 15 seconds without activity this is thrown..
-			if occurence == 3 {
-				log.Log.Info("Main: Restarting machinery.")
-				communication.HandleBootstrap <- "restart"
-				time.Sleep(2 * time.Second)
-				occurence = 0
-			}
-			previousPacket = packageCounter.Load().(int64)
-
 			time.Sleep(5 * time.Second)
 		}
 	}()

machinery/src/routers/mqtt/main.go

@@ -13,10 +13,38 @@ import (
 	"github.com/kerberos-io/agent/machinery/src/webrtc"
 )
 
+// We'll cache the MQTT settings to know if we need to reinitialize the MQTT client connection.
+// If we update the configuration but no new MQTT settings are provided, we don't need to restart it.
+var PREV_MQTTURI string
+var PREV_MQTTUsername string
+var PREV_MQTTPassword string
+var PREV_HubKey string
+var PREV_AgentKey string
+
+func HasMQTTClientModified(configuration *models.Configuration) bool {
+	MTTURI := configuration.Config.MQTTURI
+	MTTUsername := configuration.Config.MQTTUsername
+	MQTTPassword := configuration.Config.MQTTPassword
+	HubKey := configuration.Config.HubKey
+	AgentKey := configuration.Config.Key
+	if PREV_MQTTURI != MTTURI || PREV_MQTTUsername != MTTUsername || PREV_MQTTPassword != MQTTPassword || PREV_HubKey != HubKey || PREV_AgentKey != AgentKey {
+		log.Log.Info("HasMQTTClientModified: MQTT settings have been modified, restarting MQTT client.")
+		return true
+	}
+	return false
+}
+
 func ConfigureMQTT(configuration *models.Configuration, communication *models.Communication) mqtt.Client {
 
 	config := configuration.Config
 
+	// Set the MQTT settings.
+	PREV_MQTTURI = configuration.Config.MQTTURI
+	PREV_MQTTUsername = configuration.Config.MQTTUsername
+	PREV_MQTTPassword = configuration.Config.MQTTPassword
+	PREV_HubKey = configuration.Config.HubKey
+	PREV_AgentKey = configuration.Config.Key
+
 	if config.Offline == "true" {
 		log.Log.Info("ConfigureMQTT: not starting as running in Offline mode.")
 	} else {
@@ -78,7 +106,6 @@ func ConfigureMQTT(configuration *models.Configuration, communication *models.Co
 			webrtc.CandidateArrays = make(map[string](chan string))
 
 			opts.OnConnect = func(c mqtt.Client) {
-
 				// We managed to connect to the MQTT broker, hurray!
 				log.Log.Info("ConfigureMQTT: " + mqttClientID + " connected to " + mqttURL)
 
@@ -117,11 +144,15 @@ func MQTTListenerHandleLiveSD(mqttClient mqtt.Client, hubKey string, configurati
 	config := configuration.Config
 	topicRequest := "kerberos/" + hubKey + "/device/" + config.Key + "/request-live"
 	mqttClient.Subscribe(topicRequest, 0, func(c mqtt.Client, msg mqtt.Message) {
-		select {
-		case communication.HandleLiveSD <- time.Now().Unix():
-		default:
+		if communication.CameraConnected {
+			select {
+			case communication.HandleLiveSD <- time.Now().Unix():
+			default:
+			}
+			log.Log.Info("MQTTListenerHandleLiveSD: received request to livestream.")
+		} else {
+			log.Log.Info("MQTTListenerHandleLiveSD: received request to livestream, but camera is not connected.")
 		}
-		log.Log.Info("MQTTListenerHandleLiveSD: received request to livestream.")
 		msg.Ack()
 	})
 }
@@ -130,12 +161,16 @@ func MQTTListenerHandleLiveHDHandshake(mqttClient mqtt.Client, hubKey string, co
 	config := configuration.Config
 	topicRequestWebRtc := config.Key + "/register"
 	mqttClient.Subscribe(topicRequestWebRtc, 0, func(c mqtt.Client, msg mqtt.Message) {
-		log.Log.Info("MQTTListenerHandleLiveHDHandshake: received request to setup webrtc.")
-		var sdp models.SDPPayload
-		json.Unmarshal(msg.Payload(), &sdp)
-		select {
-		case communication.HandleLiveHDHandshake <- sdp:
-		default:
+		if communication.CameraConnected {
+			var sdp models.SDPPayload
+			json.Unmarshal(msg.Payload(), &sdp)
+			select {
+			case communication.HandleLiveHDHandshake <- sdp:
+			default:
+			}
+			log.Log.Info("MQTTListenerHandleLiveHDHandshake: received request to setup webrtc.")
+		} else {
+			log.Log.Info("MQTTListenerHandleLiveHDHandshake: received request to setup webrtc, but camera is not connected.")
 		}
 		msg.Ack()
 	})
@@ -145,9 +180,13 @@ func MQTTListenerHandleLiveHDKeepalive(mqttClient mqtt.Client, hubKey string, co
 	config := configuration.Config
 	topicKeepAlive := fmt.Sprintf("kerberos/webrtc/keepalivehub/%s", config.Key)
 	mqttClient.Subscribe(topicKeepAlive, 0, func(c mqtt.Client, msg mqtt.Message) {
-		alive := string(msg.Payload())
-		communication.HandleLiveHDKeepalive <- alive
-		log.Log.Info("MQTTListenerHandleLiveHDKeepalive: Received keepalive: " + alive)
+		if communication.CameraConnected {
+			alive := string(msg.Payload())
+			communication.HandleLiveHDKeepalive <- alive
+			log.Log.Info("MQTTListenerHandleLiveHDKeepalive: Received keepalive: " + alive)
+		} else {
+			log.Log.Info("MQTTListenerHandleLiveHDKeepalive: received keepalive, but camera is not connected.")
+		}
 	})
 }
 
@@ -155,9 +194,13 @@ func MQTTListenerHandleLiveHDPeers(mqttClient mqtt.Client, hubKey string, config
 	config := configuration.Config
 	topicPeers := fmt.Sprintf("kerberos/webrtc/peers/%s", config.Key)
 	mqttClient.Subscribe(topicPeers, 0, func(c mqtt.Client, msg mqtt.Message) {
-		peerCount := string(msg.Payload())
-		communication.HandleLiveHDPeers <- peerCount
-		log.Log.Info("MQTTListenerHandleLiveHDPeers: Number of peers listening: " + peerCount)
+		if communication.CameraConnected {
+			peerCount := string(msg.Payload())
+			communication.HandleLiveHDPeers <- peerCount
+			log.Log.Info("MQTTListenerHandleLiveHDPeers: Number of peers listening: " + peerCount)
+		} else {
+			log.Log.Info("MQTTListenerHandleLiveHDPeers: received peer count, but camera is not connected.")
+		}
 	})
 }
 
@@ -165,19 +208,23 @@ func MQTTListenerHandleLiveHDCandidates(mqttClient mqtt.Client, hubKey string, c
 	config := configuration.Config
 	topicCandidates := "candidate/cloud"
 	mqttClient.Subscribe(topicCandidates, 0, func(c mqtt.Client, msg mqtt.Message) {
-		var candidate models.Candidate
-		json.Unmarshal(msg.Payload(), &candidate)
-		if candidate.CloudKey == config.Key {
-			key := candidate.CloudKey + "/" + candidate.Cuuid
-			candidatesExists := false
-			var channel chan string
-			for !candidatesExists {
-				webrtc.CandidatesMutex.Lock()
-				channel, candidatesExists = webrtc.CandidateArrays[key]
-				webrtc.CandidatesMutex.Unlock()
+		if communication.CameraConnected {
+			var candidate models.Candidate
+			json.Unmarshal(msg.Payload(), &candidate)
+			if candidate.CloudKey == config.Key {
+				key := candidate.CloudKey + "/" + candidate.Cuuid
+				candidatesExists := false
+				var channel chan string
+				for !candidatesExists {
+					webrtc.CandidatesMutex.Lock()
+					channel, candidatesExists = webrtc.CandidateArrays[key]
+					webrtc.CandidatesMutex.Unlock()
+				}
+				log.Log.Info("MQTTListenerHandleLiveHDCandidates: " + string(msg.Payload()))
+				channel <- string(msg.Payload())
 			}
-			log.Log.Info("MQTTListenerHandleLiveHDCandidates: " + string(msg.Payload()))
-			channel <- string(msg.Payload())
+		} else {
+			log.Log.Info("MQTTListenerHandleLiveHDCandidates: received candidate, but camera is not connected.")
 		}
 	})
 }
@@ -186,23 +233,28 @@ func MQTTListenerHandleONVIF(mqttClient mqtt.Client, hubKey string, configuratio
 	config := configuration.Config
 	topicOnvif := fmt.Sprintf("kerberos/onvif/%s", config.Key)
 	mqttClient.Subscribe(topicOnvif, 0, func(c mqtt.Client, msg mqtt.Message) {
-		var onvifAction models.OnvifAction
-		json.Unmarshal(msg.Payload(), &onvifAction)
-		communication.HandleONVIF <- onvifAction
-		log.Log.Info("MQTTListenerHandleONVIF: Received an action - " + onvifAction.Action)
+		if communication.CameraConnected {
+			var onvifAction models.OnvifAction
+			json.Unmarshal(msg.Payload(), &onvifAction)
+			communication.HandleONVIF <- onvifAction
+			log.Log.Info("MQTTListenerHandleONVIF: Received an action - " + onvifAction.Action)
+		} else {
+			log.Log.Info("MQTTListenerHandleONVIF: received action, but camera is not connected.")
+		}
 	})
 }
 
 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)
+		// Cleanup all subscriptions
+		mqttClient.Unsubscribe("kerberos/" + PREV_HubKey + "/device/" + PREV_AgentKey + "/request-live")
+		mqttClient.Unsubscribe(PREV_AgentKey + "/register")
+		mqttClient.Unsubscribe("kerberos/webrtc/keepalivehub/" + PREV_AgentKey)
+		mqttClient.Unsubscribe("kerberos/webrtc/peers/" + PREV_AgentKey)
 		mqttClient.Unsubscribe("candidate/cloud")
-		mqttClient.Unsubscribe("kerberos/onvif/" + config.Key)
+		mqttClient.Unsubscribe("kerberos/onvif/" + PREV_AgentKey)
 		mqttClient.Disconnect(1000)
 		mqttClient = nil
+		log.Log.Info("DisconnectMQTT: MQTT client disconnected.")
 	}
 }