kamen
/
graphicek
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

MQTT Sensors (#1) 

Experimental support for MQTT based sensors

Reviewed-on: #1
Co-authored-by: Jan Zípek <jan@zipek.cz>
Co-committed-by: Jan Zípek <jan@zipek.cz>
This commit is contained in:
Jan Zípek 2025-03-05 11:10:09 +01:00 committed by kamen
parent 40fac3a6f3
commit a701173476
11 changed files with 356 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
client/src/api/sensors.ts
View File

@ -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' },

58
client/src/pages/sensors/components/SensorFormModal.tsx
View File

@ -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

5
server/app/mqtt.go Normal file
View File

@ -0,0 +1,5 @@
package app
func (s *Server) EnsureMqttListeners() {
s.Services.MQTTBrokers.EnsureListeners()
}

8
server/app/server.go
View File

@ -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
}

4
server/database/migrations/1741085898207_sensors_mqtt.sql Normal file
View File

@ -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;

51
server/integrations/mqtt.go
View File

@ -6,6 +6,17 @@ import (
type MQTTIntegration struct{}
type MQTTEvent struct {
Type string
Topic string
Payload []byte
}
type MQTTListeningClient struct {
Client MQTT.Client
Channel chan MQTTEvent
}
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 +36,43 @@ 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 MQTTEvent)
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 <- MQTTEvent{
Type: "message",
Topic: msg.Topic(),
Payload: 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
}
func (c *MQTTListeningClient) Close() {
c.Channel <- MQTTEvent{
Type: "close",
}
c.Client.Disconnect(250)
close(c.Channel)
}

5
server/main.go
View File

@ -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()
@ -116,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)
}

12
server/models/sensors.go
View File

@ -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"`
}

10
server/routes/sensors.go
View File

@ -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)

166
server/services/mqtt_brokers_service.go
View File

@ -1,9 +1,19 @@
package services
import "basic-sensor-receiver/models"
import (
"basic-sensor-receiver/integrations"
"basic-sensor-receiver/models"
"encoding/json"
"fmt"
"log"
"sync"
)
type MQTTBrokersService struct {
ctx *Context
MqttWaitGroup *sync.WaitGroup
MqttClients []*integrations.MQTTListeningClient
}
func (s *MQTTBrokersService) GetList() ([]models.MQTTBrokerItem, error) {
@ -69,6 +79,8 @@ func (s *MQTTBrokersService) Create(name string, address string, username *strin
broker.Id = int(id)
s.EnsureListeners()
return &broker, nil
}
@ -95,6 +107,8 @@ func (s *MQTTBrokersService) Update(id int64, name string, address string, usern
return nil, err
}
s.EnsureListeners()
return &broker, nil
}
@ -111,6 +125,8 @@ func (s *MQTTBrokersService) Delete(id int64) error {
return err
}
s.EnsureListeners()
return nil
}
@ -146,3 +162,151 @@ func (s *MQTTBrokersService) PublishTopic(brokerId int64, topic string, message
message,
)
}
func (s *MQTTBrokersService) StopListeners() {
if s.MqttWaitGroup != nil {
for _, client := range s.MqttClients {
client.Close()
}
s.MqttWaitGroup.Wait()
}
}
func (s *MQTTBrokersService) EnsureListeners() {
if s.MqttWaitGroup != nil {
// Disconnect all existing clients
for _, client := range s.MqttClients {
client.Close()
}
// 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 {
log.Println(fmt.Errorf("error getting MQTT brokers: %s", err))
return
}
sensors, err := s.ctx.Services.Sensors.GetList()
if err != nil {
log.Println(fmt.Errorf("error getting sensors: %s", err))
return
}
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
}
client, err := s.ctx.Integrations.MQTT.Listen(broker.Address, *broker.Username, *broker.Password, *broker.ClientId, topics)
if err != nil {
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 unexpectedly")
break
}
if data.Type == "close" {
log.Println("MQTT channel closed by request")
break
}
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
}
}
}
s.MqttWaitGroup.Done()
}()
}
}

44
server/services/sensors_service.go
View File

@ -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, mqtt_path, 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
}