From 4f9fcbe7e03a10b7f72bab9d021e4754ea1552a0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Z=C3=ADpek?= <jan@zipek.cz>
Date: Tue, 4 Mar 2025 12:09:15 +0100
Subject: [PATCH 1/6] MQTT sensors implemented, refactoring needed

---
 client/src/api/sensors.ts                     |  21 ++-
 .../sensors/components/SensorFormModal.tsx    |  58 +++++++-
 server/app/mqtt.go                            |   5 +
 server/app/server.go                          |   8 +-
 .../migrations/1741085898207_sensors_mqtt.sql |   4 +
 server/integrations/mqtt.go                   |  32 +++++
 server/main.go                                |   3 +
 server/models/sensors.go                      |  12 +-
 server/routes/sensors.go                      |  10 +-
 server/services/mqtt_brokers_service.go       | 136 +++++++++++++++++-
 server/services/sensors_service.go            |  44 ++++--
 11 files changed, 305 insertions(+), 28 deletions(-)
 create mode 100644 server/app/mqtt.go
 create mode 100644 server/database/migrations/1741085898207_sensors_mqtt.sql

diff --git a/client/src/api/sensors.ts b/client/src/api/sensors.ts
index 7252e26..9f75bc9 100644
--- a/client/src/api/sensors.ts
+++ b/client/src/api/sensors.ts
@@ -5,18 +5,33 @@ export type SensorInfo = {
 	name: string
 	authKey: string
 	lastContactAt?: number
+	type: 'rest' | 'mqtt'
+	mqttTopic?: string
+	mqttBrokerId?: number
+	mqttPath?: string
+}
+
+export type SensorModifiableData = {
+	name: string
+	type: 'rest' | 'mqtt'
+	mqttTopic?: string
+	mqttBrokerId?: number
+	mqttPath?: string
 }
 
 export const getSensors = () => request<SensorInfo[]>('/api/sensors')
 
-export const createSensor = (name: string) =>
+export const createSensor = (data: SensorModifiableData) =>
 	request<SensorInfo>('/api/sensors', {
 		method: 'POST',
 		headers: { 'content-type': 'application/json' },
-		body: JSON.stringify({ name }),
+		body: JSON.stringify(data),
 	})
 
-export const updateSensor = ({ id, ...body }: { id: number; name: string }) =>
+export const updateSensor = ({
+	id,
+	...body
+}: { id: number } & SensorModifiableData) =>
 	request<SensorInfo>(`/api/sensors/${id}`, {
 		method: 'PUT',
 		headers: { 'content-type': 'application/json' },
diff --git a/client/src/pages/sensors/components/SensorFormModal.tsx b/client/src/pages/sensors/components/SensorFormModal.tsx
index e7b90f8..94648bc 100644
--- a/client/src/pages/sensors/components/SensorFormModal.tsx
+++ b/client/src/pages/sensors/components/SensorFormModal.tsx
@@ -1,7 +1,8 @@
+import { getMQTTBrokers } from '@/api/mqttBrokers'
 import { createSensor, SensorInfo, updateSensor } from '@/api/sensors'
 import { Modal } from '@/components/Modal'
 import { useForm } from '@/utils/hooks/useForm'
-import { useMutation, useQueryClient } from 'react-query'
+import { useMutation, useQuery, useQueryClient } from 'react-query'
 
 type Props = {
 	open: boolean
@@ -14,22 +15,36 @@ export const SensorFormModal = ({ open, onClose, sensor }: Props) => {
 	const createMutation = useMutation(createSensor)
 	const updateMutation = useMutation(updateSensor)
 
-	const { handleSubmit, register } = useForm({
+	const brokers = useQuery(['/mqtt/brokers'], getMQTTBrokers)
+
+	const { handleSubmit, register, watch } = useForm({
 		defaultValue: () => ({
 			name: sensor?.name ?? '',
+			type: sensor?.type ?? 'rest',
+			mqttBrokerId: sensor?.mqttBrokerId ?? null,
+			mqttPath: sensor?.mqttPath ?? null,
+			mqttTopic: sensor?.mqttTopic ?? null,
 		}),
 		onSubmit: async (v) => {
 			if (isLoading) {
 				return
 			}
 
+			const data = {
+				name: v.name,
+				type: v.type,
+				mqttBrokerId: v.mqttBrokerId ? +v.mqttBrokerId : undefined,
+				mqttPath: v.mqttPath ?? undefined,
+				mqttTopic: v.mqttTopic ?? undefined,
+			}
+
 			if (sensor) {
 				await updateMutation.mutateAsync({
 					id: sensor.id,
-					name: v.name,
+					...data,
 				})
 			} else {
-				await createMutation.mutateAsync(v.name)
+				await createMutation.mutateAsync(data)
 			}
 
 			queryClient.invalidateQueries(['/sensors'])
@@ -38,6 +53,8 @@ export const SensorFormModal = ({ open, onClose, sensor }: Props) => {
 		},
 	})
 
+	const type = watch('type')
+
 	const isLoading = createMutation.isLoading || updateMutation.isLoading
 
 	return (
@@ -54,6 +71,39 @@ export const SensorFormModal = ({ open, onClose, sensor }: Props) => {
 					/>
 				</div>
 
+				<div className="input">
+					<label>Type</label>
+					<select {...register('type')}>
+						<option value="rest">REST</option>
+						<option value="mqtt">MQTT</option>
+					</select>
+				</div>
+
+				{type === 'mqtt' && (
+					<>
+						<div className="input">
+							<label>Broker</label>
+							<select {...register('mqttBrokerId')}>
+								{brokers.data?.map((b) => (
+									<option key={b.id} value={b.id}>
+										{b.name}
+									</option>
+								))}
+							</select>
+						</div>
+
+						<div className="input">
+							<label>Topic</label>
+							<input type="text" {...register('mqttTopic')} />
+						</div>
+
+						<div className="input">
+							<label>Path</label>
+							<input type="text" {...register('mqttPath')} />
+						</div>
+					</>
+				)}
+
 				<div className="actions">
 					<button className="cancel" type="button" onClick={onClose}>
 						Cancel
diff --git a/server/app/mqtt.go b/server/app/mqtt.go
new file mode 100644
index 0000000..4ea8b6d
--- /dev/null
+++ b/server/app/mqtt.go
@@ -0,0 +1,5 @@
+package app
+
+func (s *Server) EnsureMqttListeners() {
+	s.Services.MQTTBrokers.EnsureListeners()
+}
diff --git a/server/app/server.go b/server/app/server.go
index 35eb54d..c0a1976 100644
--- a/server/app/server.go
+++ b/server/app/server.go
@@ -9,9 +9,10 @@ import (
 )
 
 type Server struct {
-	DB       *sqlx.DB
-	Config   *config.Config
-	Services *services.Services
+	DB           *sqlx.DB
+	Config       *config.Config
+	Services     *services.Services
+	Integrations *services.Integrations
 }
 
 func InitializeServer() *Server {
@@ -29,6 +30,7 @@ func InitializeServer() *Server {
 	ctx := services.Context{DB: server.DB, Config: server.Config}
 
 	server.Services = services.InitializeServices(&ctx)
+	server.Integrations = ctx.Integrations
 
 	return &server
 }
diff --git a/server/database/migrations/1741085898207_sensors_mqtt.sql b/server/database/migrations/1741085898207_sensors_mqtt.sql
new file mode 100644
index 0000000..d0e6acc
--- /dev/null
+++ b/server/database/migrations/1741085898207_sensors_mqtt.sql
@@ -0,0 +1,4 @@
+ALTER TABLE sensors ADD COLUMN type TEXT;
+ALTER TABLE sensors ADD COLUMN mqtt_broker_id INTEGER;
+ALTER TABLE sensors ADD COLUMN mqtt_topic TEXT;
+ALTER TABLE sensors ADD COLUMN mqtt_path TEXT;
diff --git a/server/integrations/mqtt.go b/server/integrations/mqtt.go
index cce4b18..d231892 100644
--- a/server/integrations/mqtt.go
+++ b/server/integrations/mqtt.go
@@ -6,6 +6,11 @@ import (
 
 type MQTTIntegration struct{}
 
+type MQTTListeningClient struct {
+	Client  MQTT.Client
+	Channel chan [2]string
+}
+
 func (s *MQTTIntegration) Publish(server string, username string, password string, clientId string, retain bool, qos byte, topic string, message string) error {
 	opts := MQTT.NewClientOptions()
 	opts.AddBroker(server)
@@ -25,3 +30,30 @@ func (s *MQTTIntegration) Publish(server string, username string, password strin
 
 	return nil
 }
+
+func (s *MQTTIntegration) Listen(server string, username string, password string, clientId string, topics map[string]byte) (*MQTTListeningClient, error) {
+	channel := make(chan [2]string)
+
+	opts := MQTT.NewClientOptions()
+	opts.AddBroker(server)
+	opts.SetUsername(username)
+	opts.SetPassword(password)
+	opts.SetClientID(clientId)
+	opts.SetDefaultPublishHandler(func(client MQTT.Client, msg MQTT.Message) {
+		channel <- [2]string{msg.Topic(), string(msg.Payload())}
+	})
+
+	client := MQTT.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		return nil, token.Error()
+	}
+
+	if token := client.SubscribeMultiple(topics, nil); token.Wait() && token.Error() != nil {
+		return nil, token.Error()
+	}
+
+	return &MQTTListeningClient{
+		Client:  client,
+		Channel: channel,
+	}, nil
+}
diff --git a/server/main.go b/server/main.go
index 53a1389..3118adb 100644
--- a/server/main.go
+++ b/server/main.go
@@ -90,6 +90,9 @@ func main() {
 	// Starts alerts handling goroutine
 	server.StartAlerts()
 
+	// MQTT listeners
+	server.EnsureMqttListeners()
+
 	// Graceful shutdown using SIGTERM or SIGINT
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
diff --git a/server/models/sensors.go b/server/models/sensors.go
index d15cc3a..1f18dd0 100644
--- a/server/models/sensors.go
+++ b/server/models/sensors.go
@@ -1,8 +1,12 @@
 package models
 
 type SensorItem struct {
-	Id            int64  `json:"id" db:"id"`
-	Name          string `json:"name" db:"name"`
-	AuthKey       string `json:"authKey" db:"auth_key"`
-	LastContactAt *int64 `json:"lastContactAt" db:"last_contact_at"`
+	Id            int64   `json:"id" db:"id"`
+	Name          string  `json:"name" db:"name"`
+	Type          *string `json:"type" db:"type"`
+	MqttBrokerId  *int    `json:"mqttBrokerId" db:"mqtt_broker_id"`
+	MqttTopic     *string `json:"mqttTopic" db:"mqtt_topic"`
+	MqttPath      *string `json:"mqttPath" db:"mqtt_path"`
+	AuthKey       string  `json:"authKey" db:"auth_key"`
+	LastContactAt *int64  `json:"lastContactAt" db:"last_contact_at"`
 }
diff --git a/server/routes/sensors.go b/server/routes/sensors.go
index d0aa1cc..23e3c9c 100644
--- a/server/routes/sensors.go
+++ b/server/routes/sensors.go
@@ -9,7 +9,11 @@ import (
 )
 
 type postOrPutSensorsBody struct {
-	Name string `json:"name" binding:"required"`
+	Name         string  `json:"name" binding:"required"`
+	Type         *string `json:"type"`
+	MqttBrokerId *int    `json:"mqttBrokerId"`
+	MqttTopic    *string `json:"mqttTopic"`
+	MqttPath     *string `json:"mqttPath"`
 }
 
 func GetSensors(s *app.Server) gin.HandlerFunc {
@@ -33,7 +37,7 @@ func PostSensors(s *app.Server) gin.HandlerFunc {
 			return
 		}
 
-		sensor, err := s.Services.Sensors.Create(body.Name)
+		sensor, err := s.Services.Sensors.Create(body.Name, body.Type, body.MqttBrokerId, body.MqttTopic, body.MqttPath)
 
 		if err != nil {
 			c.AbortWithError(http.StatusInternalServerError, err)
@@ -56,7 +60,7 @@ func PutSensor(s *app.Server) gin.HandlerFunc {
 			return
 		}
 
-		sensor, err := s.Services.Sensors.Update(sensorId, body.Name)
+		sensor, err := s.Services.Sensors.Update(sensorId, body.Name, body.Type, body.MqttBrokerId, body.MqttTopic, body.MqttPath)
 
 		if err != nil {
 			c.AbortWithError(http.StatusInternalServerError, err)
diff --git a/server/services/mqtt_brokers_service.go b/server/services/mqtt_brokers_service.go
index 3edefca..e675299 100644
--- a/server/services/mqtt_brokers_service.go
+++ b/server/services/mqtt_brokers_service.go
@@ -1,9 +1,18 @@
 package services
 
-import "basic-sensor-receiver/models"
+import (
+	"basic-sensor-receiver/integrations"
+	"basic-sensor-receiver/models"
+	"encoding/json"
+	"fmt"
+	"sync"
+)
 
 type MQTTBrokersService struct {
 	ctx *Context
+
+	MqttWaitGroup *sync.WaitGroup
+	MqttClients   []*integrations.MQTTListeningClient
 }
 
 func (s *MQTTBrokersService) GetList() ([]models.MQTTBrokerItem, error) {
@@ -69,6 +78,8 @@ func (s *MQTTBrokersService) Create(name string, address string, username *strin
 
 	broker.Id = int(id)
 
+	s.EnsureListeners()
+
 	return &broker, nil
 }
 
@@ -95,6 +106,8 @@ func (s *MQTTBrokersService) Update(id int64, name string, address string, usern
 		return nil, err
 	}
 
+	s.EnsureListeners()
+
 	return &broker, nil
 }
 
@@ -111,6 +124,8 @@ func (s *MQTTBrokersService) Delete(id int64) error {
 		return err
 	}
 
+	s.EnsureListeners()
+
 	return nil
 }
 
@@ -146,3 +161,122 @@ func (s *MQTTBrokersService) PublishTopic(brokerId int64, topic string, message
 		message,
 	)
 }
+
+func (s *MQTTBrokersService) EnsureListeners() {
+	if s.MqttWaitGroup != nil {
+		// Disconnect all existing clients
+		for _, client := range s.MqttClients {
+			if client.Client == nil {
+				continue
+			}
+
+			client.Client.Disconnect(250)
+		}
+
+		// Wait for all channels to clear
+		s.MqttWaitGroup.Wait()
+
+		// Clear data
+		s.MqttWaitGroup = nil
+		s.MqttClients = []*integrations.MQTTListeningClient{}
+	}
+
+	brokers, err := s.GetList()
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	sensors, err := s.ctx.Services.Sensors.GetList()
+
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+
+	// TODO: This needs to be refreshed when a new broker is added or a broker is updated
+	// TODO: Save all the coroutines and stop them when the server is stopped
+
+	s.MqttWaitGroup = &sync.WaitGroup{}
+
+	for _, broker := range brokers {
+		if broker.Username == nil {
+			broker.Username = new(string)
+		}
+
+		if broker.Password == nil {
+			broker.Password = new(string)
+		}
+
+		if broker.ClientId == nil {
+			broker.ClientId = new(string)
+		}
+
+		topics := map[string]byte{}
+		brokerSensors := []*models.SensorItem{}
+
+		for _, sensor := range sensors {
+			if sensor.MqttBrokerId == nil || *sensor.MqttBrokerId != broker.Id {
+				continue
+			}
+
+			if sensor.MqttTopic == nil {
+				continue
+			}
+
+			topics[*sensor.MqttTopic] = byte(0)
+			brokerSensors = append(brokerSensors, &sensor)
+		}
+
+		if len(brokerSensors) == 0 {
+			continue
+		}
+
+		fmt.Print("LISTENING TO TOPICS: ")
+		for topic := range topics {
+			fmt.Printf("%s ", topic)
+		}
+		fmt.Println()
+
+		client, err := s.ctx.Integrations.MQTT.Listen(broker.Address, *broker.Username, *broker.Password, *broker.ClientId, topics)
+
+		if err != nil {
+			fmt.Println(err)
+			continue
+		}
+
+		s.MqttClients = append(s.MqttClients, client)
+		s.MqttWaitGroup.Add(1)
+
+		go func() {
+			for {
+				data, ok := <-client.Channel
+				if !ok {
+					break
+				}
+
+				fmt.Printf("RECEIVED TOPIC: %s MESSAGE: %s\n", data[0], data[1])
+
+				// TODO: This is sub-optional
+				for _, sensor := range brokerSensors {
+					if sensor.MqttTopic == nil || *sensor.MqttTopic != data[0] || sensor.MqttPath == nil {
+						continue
+					}
+
+					var responseJson map[string]any
+					err := json.Unmarshal([]byte(data[1]), &responseJson)
+
+					if err != nil {
+						fmt.Print(err)
+						continue
+					}
+
+					s.ctx.Services.SensorValues.Push(sensor.Id, responseJson[*sensor.MqttPath].(float64))
+				}
+			}
+
+			s.MqttWaitGroup.Done()
+		}()
+	}
+}
diff --git a/server/services/sensors_service.go b/server/services/sensors_service.go
index 5fd14ce..1d1e28e 100644
--- a/server/services/sensors_service.go
+++ b/server/services/sensors_service.go
@@ -13,7 +13,7 @@ type SensorsService struct {
 func (s *SensorsService) GetList() ([]models.SensorItem, error) {
 	sensors := []models.SensorItem{}
 
-	err := s.ctx.DB.Select(&sensors, "SELECT id, name, auth_key, last_contact_at FROM sensors")
+	err := s.ctx.DB.Select(&sensors, "SELECT id, name, type, mqtt_broker_id, mqtt_topic, auth_key, last_contact_at FROM sensors")
 
 	if err != nil {
 		return nil, err
@@ -22,19 +22,28 @@ func (s *SensorsService) GetList() ([]models.SensorItem, error) {
 	return sensors, nil
 }
 
-func (s *SensorsService) Create(name string) (*models.SensorItem, error) {
+func (s *SensorsService) Create(name string, sensorType *string, mqttBrokerId *int, mqttTopic *string, mqttPath *string) (*models.SensorItem, error) {
 	authKey, err := generateRandomString(32)
 
 	if err != nil {
 		return nil, err
 	}
 
-	item := models.SensorItem{
-		Name:    name,
-		AuthKey: authKey,
+	if sensorType == nil {
+		sensorType = new(string)
+		*sensorType = "rest"
 	}
 
-	res, err := s.ctx.DB.Exec("INSERT INTO sensors (name, auth_key) VALUES (?, ?)", item.Name, item.AuthKey)
+	item := models.SensorItem{
+		Name:         name,
+		AuthKey:      authKey,
+		Type:         sensorType,
+		MqttBrokerId: mqttBrokerId,
+		MqttTopic:    mqttTopic,
+		MqttPath:     mqttPath,
+	}
+
+	res, err := s.ctx.DB.Exec("INSERT INTO sensors (name, auth_key, type, mqtt_broker_id, mqtt_topic, mqtt_path) VALUES (?, ?, ?, ?, ?, ?)", item.Name, item.AuthKey, item.Type, item.MqttBrokerId, item.MqttTopic, item.MqttPath)
 
 	if err != nil {
 		return nil, err
@@ -46,13 +55,15 @@ func (s *SensorsService) Create(name string) (*models.SensorItem, error) {
 		return nil, err
 	}
 
+	s.ctx.Services.MQTTBrokers.EnsureListeners()
+
 	return &item, nil
 }
 
 func (s *SensorsService) GetById(id int64) (*models.SensorItem, error) {
 	item := models.SensorItem{}
 
-	err := s.ctx.DB.Get(&item, "SELECT id, name, auth_key, last_contact_at FROM sensors WHERE id = $1", id)
+	err := s.ctx.DB.Get(&item, "SELECT id, name, auth_key, last_contact_at, type, mqtt_broker_id, mqtt_topic, mqtt_path FROM sensors WHERE id = $1", id)
 
 	if err != nil {
 		return nil, err
@@ -61,21 +72,32 @@ func (s *SensorsService) GetById(id int64) (*models.SensorItem, error) {
 	return &item, nil
 }
 
-func (s *SensorsService) Update(id int64, name string) (*models.SensorItem, error) {
+func (s *SensorsService) Update(id int64, name string, sensorType *string, mqttBrokerId *int, mqttTopic *string, mqttPath *string) (*models.SensorItem, error) {
 	item, err := s.GetById(id)
 
 	if err != nil {
 		return nil, err
 	}
 
-	item.Name = name
+	if sensorType == nil {
+		sensorType = new(string)
+		*sensorType = "rest"
+	}
 
-	_, err = s.ctx.DB.Exec("UPDATE sensors SET name = ? WHERE id = ?", item.Name, item.Id)
+	item.Name = name
+	item.Type = sensorType
+	item.MqttBrokerId = mqttBrokerId
+	item.MqttTopic = mqttTopic
+	item.MqttPath = mqttPath
+
+	_, err = s.ctx.DB.Exec("UPDATE sensors SET name = ?, type = ?, mqtt_broker_id = ?, mqtt_topic = ?, mqtt_path = ? WHERE id = ?", item.Name, item.Type, item.MqttBrokerId, item.MqttTopic, item.MqttPath, item.Id)
 
 	if err != nil {
 		return nil, err
 	}
 
+	s.ctx.Services.MQTTBrokers.EnsureListeners()
+
 	return item, nil
 }
 
@@ -94,6 +116,8 @@ func (s *SensorsService) DeleteById(id int64) error {
 
 	_, err = s.ctx.DB.Exec("DELETE FROM sensors WHERE id = ?", item.Id)
 
+	s.ctx.Services.MQTTBrokers.EnsureListeners()
+
 	return err
 }
 
-- 
2.40.1


From 9f65112a821ee0d83540f0639be3c74b15c6ed88 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Z=C3=ADpek?= <jan@zipek.cz>
Date: Tue, 4 Mar 2025 21:31:42 +0100
Subject: [PATCH 2/6] Fix missing value field

---
 server/services/mqtt_brokers_service.go | 7 ++++++-
 server/services/sensors_service.go      | 2 +-
 2 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/server/services/mqtt_brokers_service.go b/server/services/mqtt_brokers_service.go
index e675299..2fa2bee 100644
--- a/server/services/mqtt_brokers_service.go
+++ b/server/services/mqtt_brokers_service.go
@@ -260,7 +260,12 @@ func (s *MQTTBrokersService) EnsureListeners() {
 
 				// TODO: This is sub-optional
 				for _, sensor := range brokerSensors {
-					if sensor.MqttTopic == nil || *sensor.MqttTopic != data[0] || sensor.MqttPath == nil {
+					if sensor.MqttTopic == nil || sensor.MqttPath == nil {
+						fmt.Printf("Skipping %s because it's missing topic or path", sensor.Name)
+						continue
+					}
+
+					if *sensor.MqttTopic != data[0] {
 						continue
 					}
 
diff --git a/server/services/sensors_service.go b/server/services/sensors_service.go
index 1d1e28e..63f4e63 100644
--- a/server/services/sensors_service.go
+++ b/server/services/sensors_service.go
@@ -13,7 +13,7 @@ type SensorsService struct {
 func (s *SensorsService) GetList() ([]models.SensorItem, error) {
 	sensors := []models.SensorItem{}
 
-	err := s.ctx.DB.Select(&sensors, "SELECT id, name, type, mqtt_broker_id, mqtt_topic, auth_key, last_contact_at FROM sensors")
+	err := s.ctx.DB.Select(&sensors, "SELECT id, name, type, mqtt_broker_id, mqtt_topic, mqtt_path, auth_key, last_contact_at FROM sensors")
 
 	if err != nil {
 		return nil, err
-- 
2.40.1


From 3029d4af52ec97977d28a44ca04bf39045912796 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Z=C3=ADpek?= <jan@zipek.cz>
Date: Tue, 4 Mar 2025 23:29:48 +0100
Subject: [PATCH 3/6] Improved error messages

---
 server/services/mqtt_brokers_service.go | 30 +++++++++++++++----------
 1 file changed, 18 insertions(+), 12 deletions(-)

diff --git a/server/services/mqtt_brokers_service.go b/server/services/mqtt_brokers_service.go
index 2fa2bee..177c894 100644
--- a/server/services/mqtt_brokers_service.go
+++ b/server/services/mqtt_brokers_service.go
@@ -184,14 +184,14 @@ func (s *MQTTBrokersService) EnsureListeners() {
 	brokers, err := s.GetList()
 
 	if err != nil {
-		fmt.Println(err)
+		fmt.Println(fmt.Errorf("error getting MQTT brokers: %s", err))
 		return
 	}
 
 	sensors, err := s.ctx.Services.Sensors.GetList()
 
 	if err != nil {
-		fmt.Println(err)
+		fmt.Println(fmt.Errorf("error getting sensors: %s", err))
 		return
 	}
 
@@ -233,16 +233,10 @@ func (s *MQTTBrokersService) EnsureListeners() {
 			continue
 		}
 
-		fmt.Print("LISTENING TO TOPICS: ")
-		for topic := range topics {
-			fmt.Printf("%s ", topic)
-		}
-		fmt.Println()
-
 		client, err := s.ctx.Integrations.MQTT.Listen(broker.Address, *broker.Username, *broker.Password, *broker.ClientId, topics)
 
 		if err != nil {
-			fmt.Println(err)
+			fmt.Println(fmt.Errorf("error listening to MQTT broker %s: %s", broker.Name, err))
 			continue
 		}
 
@@ -261,7 +255,7 @@ func (s *MQTTBrokersService) EnsureListeners() {
 				// TODO: This is sub-optional
 				for _, sensor := range brokerSensors {
 					if sensor.MqttTopic == nil || sensor.MqttPath == nil {
-						fmt.Printf("Skipping %s because it's missing topic or path", sensor.Name)
+						fmt.Printf("WARN: Skipping sensor %s because it's missing topic or value path", sensor.Name)
 						continue
 					}
 
@@ -273,11 +267,23 @@ func (s *MQTTBrokersService) EnsureListeners() {
 					err := json.Unmarshal([]byte(data[1]), &responseJson)
 
 					if err != nil {
-						fmt.Print(err)
+						fmt.Print(fmt.Errorf("error parsing MQTT message JSON: %s", err))
 						continue
 					}
 
-					s.ctx.Services.SensorValues.Push(sensor.Id, responseJson[*sensor.MqttPath].(float64))
+					if _, exists := responseJson[*sensor.MqttPath]; !exists {
+						fmt.Printf("WARN: Skipping sensor %s because it's value path %s was not found in the message", sensor.Name, *sensor.MqttPath)
+						continue
+					}
+
+					value, ok := responseJson[*sensor.MqttPath].(float64)
+
+					if !ok {
+						fmt.Printf("WARN: Skipping sensor %s because value path %s doesn't resolve to number", sensor.Name, *sensor.MqttPath)
+						continue
+					}
+
+					s.ctx.Services.SensorValues.Push(sensor.Id, value)
 				}
 			}
 
-- 
2.40.1


From bae535a014d4b22a4c425b211814ac1668fb1da6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Z=C3=ADpek?= <jan@zipek.cz>
Date: Wed, 5 Mar 2025 11:00:51 +0100
Subject: [PATCH 4/6] Refactoring

---
 server/integrations/mqtt.go             |  25 +++++-
 server/main.go                          |   2 +
 server/services/mqtt_brokers_service.go | 100 +++++++++++++++---------
 3 files changed, 89 insertions(+), 38 deletions(-)

diff --git a/server/integrations/mqtt.go b/server/integrations/mqtt.go
index d231892..166d35d 100644
--- a/server/integrations/mqtt.go
+++ b/server/integrations/mqtt.go
@@ -6,9 +6,15 @@ import (
 
 type MQTTIntegration struct{}
 
+type MQTTEvent struct {
+	Type    string
+	Topic   string
+	Payload []byte
+}
+
 type MQTTListeningClient struct {
 	Client  MQTT.Client
-	Channel chan [2]string
+	Channel chan MQTTEvent
 }
 
 func (s *MQTTIntegration) Publish(server string, username string, password string, clientId string, retain bool, qos byte, topic string, message string) error {
@@ -32,7 +38,7 @@ func (s *MQTTIntegration) Publish(server string, username string, password strin
 }
 
 func (s *MQTTIntegration) Listen(server string, username string, password string, clientId string, topics map[string]byte) (*MQTTListeningClient, error) {
-	channel := make(chan [2]string)
+	channel := make(chan MQTTEvent)
 
 	opts := MQTT.NewClientOptions()
 	opts.AddBroker(server)
@@ -40,7 +46,11 @@ func (s *MQTTIntegration) Listen(server string, username string, password string
 	opts.SetPassword(password)
 	opts.SetClientID(clientId)
 	opts.SetDefaultPublishHandler(func(client MQTT.Client, msg MQTT.Message) {
-		channel <- [2]string{msg.Topic(), string(msg.Payload())}
+		channel <- MQTTEvent{
+			Type:    "message",
+			Topic:   msg.Topic(),
+			Payload: msg.Payload(),
+		}
 	})
 
 	client := MQTT.NewClient(opts)
@@ -57,3 +67,12 @@ func (s *MQTTIntegration) Listen(server string, username string, password string
 		Channel: channel,
 	}, nil
 }
+
+func (c *MQTTListeningClient) Close() {
+	c.Channel <- MQTTEvent{
+		Type: "close",
+	}
+
+	c.Client.Disconnect(250)
+	close(c.Channel)
+}
diff --git a/server/main.go b/server/main.go
index 3118adb..43c0de5 100644
--- a/server/main.go
+++ b/server/main.go
@@ -119,6 +119,8 @@ func main() {
 	// Shutdown the server
 	log.Println("Shutting down server...")
 
+	server.Services.MQTTBrokers.StopListeners()
+
 	if err := srv.Shutdown(context.TODO()); err != nil {
 		log.Fatalf("Server shutdown failed: %v", err)
 	}
diff --git a/server/services/mqtt_brokers_service.go b/server/services/mqtt_brokers_service.go
index 177c894..d674bf8 100644
--- a/server/services/mqtt_brokers_service.go
+++ b/server/services/mqtt_brokers_service.go
@@ -5,6 +5,7 @@ import (
 	"basic-sensor-receiver/models"
 	"encoding/json"
 	"fmt"
+	"log"
 	"sync"
 )
 
@@ -162,6 +163,20 @@ func (s *MQTTBrokersService) PublishTopic(brokerId int64, topic string, message
 	)
 }
 
+func (s *MQTTBrokersService) StopListeners() {
+	if s.MqttWaitGroup != nil {
+		for _, client := range s.MqttClients {
+			if client.Client == nil {
+				continue
+			}
+
+			client.Client.Disconnect(250)
+		}
+
+		s.MqttWaitGroup.Wait()
+	}
+}
+
 func (s *MQTTBrokersService) EnsureListeners() {
 	if s.MqttWaitGroup != nil {
 		// Disconnect all existing clients
@@ -184,14 +199,14 @@ func (s *MQTTBrokersService) EnsureListeners() {
 	brokers, err := s.GetList()
 
 	if err != nil {
-		fmt.Println(fmt.Errorf("error getting MQTT brokers: %s", err))
+		log.Println(fmt.Errorf("error getting MQTT brokers: %s", err))
 		return
 	}
 
 	sensors, err := s.ctx.Services.Sensors.GetList()
 
 	if err != nil {
-		fmt.Println(fmt.Errorf("error getting sensors: %s", err))
+		log.Println(fmt.Errorf("error getting sensors: %s", err))
 		return
 	}
 
@@ -236,54 +251,69 @@ func (s *MQTTBrokersService) EnsureListeners() {
 		client, err := s.ctx.Integrations.MQTT.Listen(broker.Address, *broker.Username, *broker.Password, *broker.ClientId, topics)
 
 		if err != nil {
-			fmt.Println(fmt.Errorf("error listening to MQTT broker %s: %s", broker.Name, err))
+			log.Println(fmt.Errorf("error listening to MQTT broker %s: %s", broker.Name, err))
 			continue
 		}
 
 		s.MqttClients = append(s.MqttClients, client)
 		s.MqttWaitGroup.Add(1)
 
+		log.Printf("MQTT broker %s: Listening for %d topics\n", broker.Name, len(topics))
+
 		go func() {
 			for {
 				data, ok := <-client.Channel
 				if !ok {
+					log.Println("WARN: MQTT channel closed")
 					break
 				}
 
-				fmt.Printf("RECEIVED TOPIC: %s MESSAGE: %s\n", data[0], data[1])
+				if data.Type == "close" {
+					log.Println("MQTT channel closed by request")
+					break
+				}
 
-				// TODO: This is sub-optional
-				for _, sensor := range brokerSensors {
-					if sensor.MqttTopic == nil || sensor.MqttPath == nil {
-						fmt.Printf("WARN: Skipping sensor %s because it's missing topic or value path", sensor.Name)
-						continue
+				switch data.Type {
+				case "message":
+					{
+						log.Printf("MQTT Message: Topic: %s Message: %s\n", data.Topic, string(data.Payload))
+
+						// TODO: This is sub-optional
+						for _, sensor := range brokerSensors {
+							if sensor.MqttTopic == nil || sensor.MqttPath == nil {
+								log.Printf("WARN: Skipping sensor %s because it's missing topic or value path", sensor.Name)
+								continue
+							}
+
+							if *sensor.MqttTopic != data.Topic {
+								continue
+							}
+
+							var responseJson map[string]any
+							err := json.Unmarshal(data.Payload, &responseJson)
+
+							if err != nil {
+								log.Print(fmt.Errorf("error parsing MQTT message JSON: %s", err))
+								continue
+							}
+
+							if _, exists := responseJson[*sensor.MqttPath]; !exists {
+								log.Printf("WARN: Skipping sensor %s because it's value path %s was not found in the message", sensor.Name, *sensor.MqttPath)
+								continue
+							}
+
+							value, ok := responseJson[*sensor.MqttPath].(float64)
+
+							if !ok {
+								log.Printf("WARN: Skipping sensor %s because value path %s doesn't resolve to number", sensor.Name, *sensor.MqttPath)
+								continue
+							}
+
+							s.ctx.Services.SensorValues.Push(sensor.Id, value)
+						}
+
+						break
 					}
-
-					if *sensor.MqttTopic != data[0] {
-						continue
-					}
-
-					var responseJson map[string]any
-					err := json.Unmarshal([]byte(data[1]), &responseJson)
-
-					if err != nil {
-						fmt.Print(fmt.Errorf("error parsing MQTT message JSON: %s", err))
-						continue
-					}
-
-					if _, exists := responseJson[*sensor.MqttPath]; !exists {
-						fmt.Printf("WARN: Skipping sensor %s because it's value path %s was not found in the message", sensor.Name, *sensor.MqttPath)
-						continue
-					}
-
-					value, ok := responseJson[*sensor.MqttPath].(float64)
-
-					if !ok {
-						fmt.Printf("WARN: Skipping sensor %s because value path %s doesn't resolve to number", sensor.Name, *sensor.MqttPath)
-						continue
-					}
-
-					s.ctx.Services.SensorValues.Push(sensor.Id, value)
 				}
 			}
 
-- 
2.40.1


From 44d57143d0655d90016b0243db157bdb9715d3ff Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Z=C3=ADpek?= <jan@zipek.cz>
Date: Wed, 5 Mar 2025 11:05:24 +0100
Subject: [PATCH 5/6] Fix closing

---
 server/services/mqtt_brokers_service.go | 14 +++-----------
 1 file changed, 3 insertions(+), 11 deletions(-)

diff --git a/server/services/mqtt_brokers_service.go b/server/services/mqtt_brokers_service.go
index d674bf8..093de8b 100644
--- a/server/services/mqtt_brokers_service.go
+++ b/server/services/mqtt_brokers_service.go
@@ -166,11 +166,7 @@ func (s *MQTTBrokersService) PublishTopic(brokerId int64, topic string, message
 func (s *MQTTBrokersService) StopListeners() {
 	if s.MqttWaitGroup != nil {
 		for _, client := range s.MqttClients {
-			if client.Client == nil {
-				continue
-			}
-
-			client.Client.Disconnect(250)
+			client.Close()
 		}
 
 		s.MqttWaitGroup.Wait()
@@ -181,11 +177,7 @@ func (s *MQTTBrokersService) EnsureListeners() {
 	if s.MqttWaitGroup != nil {
 		// Disconnect all existing clients
 		for _, client := range s.MqttClients {
-			if client.Client == nil {
-				continue
-			}
-
-			client.Client.Disconnect(250)
+			client.Close()
 		}
 
 		// Wait for all channels to clear
@@ -264,7 +256,7 @@ func (s *MQTTBrokersService) EnsureListeners() {
 			for {
 				data, ok := <-client.Channel
 				if !ok {
-					log.Println("WARN: MQTT channel closed")
+					log.Println("WARN: MQTT channel closed unexpectedly")
 					break
 				}
 
-- 
2.40.1


From b6d2d115bf6dff0d5ab66424597f97639e5613b6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Z=C3=ADpek?= <jan@zipek.cz>
Date: Wed, 5 Mar 2025 11:09:33 +0100
Subject: [PATCH 6/6] TODOs are complete

---
 server/services/mqtt_brokers_service.go | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/server/services/mqtt_brokers_service.go b/server/services/mqtt_brokers_service.go
index 093de8b..6e89a3e 100644
--- a/server/services/mqtt_brokers_service.go
+++ b/server/services/mqtt_brokers_service.go
@@ -202,9 +202,6 @@ func (s *MQTTBrokersService) EnsureListeners() {
 		return
 	}
 
-	// TODO: This needs to be refreshed when a new broker is added or a broker is updated
-	// TODO: Save all the coroutines and stop them when the server is stopped
-
 	s.MqttWaitGroup = &sync.WaitGroup{}
 
 	for _, broker := range brokers {
-- 
2.40.1