imporved mqtt message handling

README.md

@@ -32,8 +32,7 @@ MQTT_HOST | localhost:1883 | Hostadresse MQTT-Broker
 MQTT_TOPIC | sensor/# | MQTT-Topic, in welchem nach Wetterdaten geschaut wird
 MQTT_USER | mqtt | Username für MQTT
 MQTT_PASS | mqtt | Passwort für MQTT
-MQTT_PUBLISH_INTERVALL | 2500 | Intervall, nachdem über MQTT empfangene Wetterdaten in die DB geschrieben werden (in Millisekunden)
+MQTT_PUBLISH_DELAY | 1000 | Innerhalb dieser Zeitspanne wird ein Wetterdatensatz noch durch weiter eintreffende Werte ergänzt. Danach wird der Datensatz veröffentlicht (in Millisekunden)
-MQTT_MIN_DIST_LAST_VALUE | 250 | Zeit, die Wetterdaten mindestens zurückgehalten werden, bevor diese in die DB geschrieben werden -> Innerhalb dieser Zeitspanne kann ein Wetterdatensatz noch durch andere Werte ergänzt werden(in Millisekunden)
 MQTT_ANONYMOUS | false | Anonyme Anmeldung am MQTT-Broker verwenden (ohne Username und Passwort)
 ALLOW_UNREGISTERED_SENSORS | false | Wetterdaten nicht registrierter Sensoren erlauben
 

config/config.go

@@ -26,8 +26,7 @@ type MqttConfig struct {
 	Topic                        string
 	Username                     string
 	Password                     string
-	PublishInterval              time.Duration
+	PublishDelay                 time.Duration
-	MinDistToLastValue           time.Duration
 	AllowAnonymousAuthentication bool
 }
 
@@ -55,8 +54,7 @@ var MqttConfiguration = MqttConfig{
 	Topic:                        getEnv("MQTT_TOPIC", "sensor/#"),
 	Username:                     getEnv("MQTT_USER", "mqtt"),
 	Password:                     getEnv("MQTT_PASS", "mqtt"),
-	PublishInterval:              getEnvDuration("MQTT_PUBLISH_INTERVALL", time.Millisecond*2500),
+	PublishDelay:                 getEnvDuration("MQTT_PUBLISH_DELAY", time.Second),
-	MinDistToLastValue:           getEnvDuration("MQTT_MIN_DIST_LAST_VALUE", time.Millisecond*250),
 	AllowAnonymousAuthentication: getEnvBool("MQTT_ANONYMOUS", false),
 }
 

run_default.ps1

@@ -11,8 +11,7 @@ Set-Item -Path "Env:MQTT_HOST" -Value "localhost:1883"
 Set-Item -Path "Env:MQTT_TOPIC" -Value "sensor/#"
 Set-Item -Path "Env:MQTT_USER" -Value "mqtt"
 Set-Item -Path "Env:MQTT_PASS" -Value "mqtt"
-Set-Item -Path "Env:MQTT_PUBLISH_INTERVALL" -Value "2500"
+Set-Item -Path "Env:MQTT_PUBLISH_DELAY" -Value "1000"
-Set-Item -Path "Env:MQTT_MIN_DIST_LAST_VALUE" -Value "250"
 Set-Item -Path "Env:MQTT_ANONYMOUS" -Value "false"
 
 Set-Item -Path "Env:MONGO_HOST" -Value "localhost:27017"

weathersource/mqtt-source.go

@@ -4,6 +4,7 @@ import (
 	"log"
 	"regexp"
 	"strconv"
+	"sync"
 	"time"
 	"weather-data/config"
 	"weather-data/storage"
@@ -16,11 +17,14 @@ var mqttTopicRegexPattern = "(^sensor)/([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F
 
 var regexTopic *regexp.Regexp = regexp.MustCompile(mqttTopicRegexPattern)
 
+var channelBufferSize = 10
+
 type mqttWeatherSource struct {
 	config                   config.MqttConfig
 	mqttClient               mqtt.Client
-	lastWeatherDataPoints []*storage.WeatherData
 	weatherSource            WeatherSourceBase
+	activeSensorMeasurements map[uuid.UUID](chan map[storage.SensorValueType]float64)
+	sensorMutex              sync.RWMutex
 }
 
 //Close mqtt client
@@ -37,7 +41,7 @@ func NewMqttSource(cfg config.MqttConfig) (*mqttWeatherSource, error) {
 
 	//mqtt
 	opts.SetKeepAlive(60 * time.Second)
-	opts.SetDefaultPublishHandler(source.mqttMessageHandler())
+	opts.SetDefaultPublishHandler(source.mqttMessageHandler)
 	opts.SetPingTimeout(1 * time.Second)
 
 	if !cfg.AllowAnonymousAuthentication {
@@ -55,16 +59,15 @@ func NewMqttSource(cfg config.MqttConfig) (*mqttWeatherSource, error) {
 		return nil, token.Error()
 	}
 
-	go source.publishDataValues()
+	source.activeSensorMeasurements = make(map[uuid.UUID]chan map[storage.SensorValueType]float64)
+	source.sensorMutex = sync.RWMutex{}
 
 	log.Print("successfully connected to mqtt-broker")
 	return source, nil
 }
 
 //mqttMessageHandler returns a function that handles incoming mqtt-messages
-func (source *mqttWeatherSource) mqttMessageHandler() mqtt.MessageHandler {
+func (source *mqttWeatherSource) mqttMessageHandler(client mqtt.Client, msg mqtt.Message) {
-
-	return func(client mqtt.Client, msg mqtt.Message) {
 	if !regexTopic.MatchString(msg.Topic()) {
 		return
 	}
@@ -74,51 +77,57 @@ func (source *mqttWeatherSource) mqttMessageHandler() mqtt.MessageHandler {
 		return
 	}
 
-		lastWeatherData, found := source.getUnwrittenDatapoints(sensorId)
-
-		if !found {
-			lastWeatherData = storage.NewWeatherData()
-			lastWeatherData.SensorId = sensorId
-			source.lastWeatherDataPoints = append(source.lastWeatherDataPoints, lastWeatherData)
-		}
-
 	value, err := strconv.ParseFloat(string(msg.Payload()), 64)
 	if err != nil {
 		return
 	}
 
 	sensorValueType := storage.SensorValueType(regexTopic.FindStringSubmatch(msg.Topic())[3])
-		lastWeatherData.Values[sensorValueType] = value
+
-		lastWeatherData.TimeStamp = time.Now()
+	dataValue := map[storage.SensorValueType]float64{
+		sensorValueType: value,
+	}
+
+	source.sensorMutex.RLock()
+	dataChannel, exists := source.activeSensorMeasurements[sensorId]
+	if !exists {
+		dataChannel = make(chan map[storage.SensorValueType]float64, channelBufferSize)
+	}
+	dataChannel <- dataValue
+	source.sensorMutex.RUnlock()
+
+	if !exists {
+		go source.publishSensorMeasurement(sensorId, dataChannel)
+		go source.cleanupSensorMeasurement(sensorId, dataChannel)
+
+		source.sensorMutex.Lock()
+		source.activeSensorMeasurements[sensorId] = dataChannel
+		source.sensorMutex.Unlock()
 	}
 }
 
-func (source *mqttWeatherSource) publishDataValues() {
+func (source *mqttWeatherSource) cleanupSensorMeasurement(sensorId uuid.UUID, channel chan<- map[storage.SensorValueType]float64) {
-	for {
+	time.Sleep(source.config.PublishDelay)
-		for len(source.lastWeatherDataPoints) != 0 {
+
-			current := *source.lastWeatherDataPoints[0]
+	source.sensorMutex.Lock()
-			diff := time.Since(current.TimeStamp)
+	delete(source.activeSensorMeasurements, sensorId)
-			if diff >= source.config.MinDistToLastValue {
+	source.sensorMutex.Unlock()
-				if err := source.newWeatherData(current); err != nil {
+
-					log.Fatal(err)
+	close(channel)
-					//if error than put the dataPoint to the end of the slice and try again later
-					dataPoint := source.lastWeatherDataPoints[0]
-					source.lastWeatherDataPoints = append(source.lastWeatherDataPoints, dataPoint)
 }
-				source.lastWeatherDataPoints = source.lastWeatherDataPoints[1:]
+
-			}
+func (source *mqttWeatherSource) publishSensorMeasurement(sensorId uuid.UUID, channel <-chan map[storage.SensorValueType]float64) {
-		}
+	weatherData := storage.NewWeatherData()
-		time.Sleep(source.config.PublishInterval)
+	weatherData.TimeStamp = time.Now()
+	weatherData.SensorId = sensorId
+
+	for values := range channel {
+		for k, v := range values {
+			weatherData.Values[k] = v
 		}
 	}
 
-func (source *mqttWeatherSource) getUnwrittenDatapoints(sensorId uuid.UUID) (*storage.WeatherData, bool) {
+	source.newWeatherData(*weatherData)
-	for _, data := range source.lastWeatherDataPoints {
-		if data.SensorId == sensorId {
-			return data, true
-		}
-	}
-	return nil, false
 }
 
 //AddNewWeatherDataCallback adds a new callbackMethod for incoming weather data