add more comments

Author: rainu

cmd/mqtt-executor/log.go

@@ -3,8 +3,10 @@ package main
 import "go.uber.org/zap"
 
 func init() {
+	//initialise our global logger
+
 	logger, _ := zap.NewDevelopment(
-		zap.AddStacktrace(zap.FatalLevel),
+		zap.AddStacktrace(zap.FatalLevel), //disable stacktrace for level lower than fatal
 	)
 	zap.ReplaceGlobals(logger)
 	defer zap.L().Sync()

cmd/mqtt-executor/main.go

@@ -24,6 +24,7 @@ func main() {
 	trigger.Executor = commandExecutor
 	sensorWorker.Executor = commandExecutor
 
+	//reacting to signals (interrupt)
 	signals := make(chan os.Signal, 1)
 	defer close(signals)
 	signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
@@ -37,6 +38,7 @@ func main() {
 		zap.L().Fatal("Error while connecting to mqtt broker: %s", zap.Error(token.Error()))
 	}
 
+	//if hassio is enabled -> publish the hassio mqtt-discovery configs
 	if *Config.HomeassistantEnable {
 		haClient := hassio.Client{
 			DeviceName:  *Config.DeviceName,
@@ -69,6 +71,8 @@ func shutdown(client MQTT.Client) {
 	}
 	closeables := []closable{&statusWorker, &sensorWorker, &trigger, commandExecutor}
 
+	//most operating systems wait a maximum of 30 seconds
+
 	wg := sync.WaitGroup{}
 	wg.Add(len(closeables))
 	timeout := 20 * time.Second
@@ -84,5 +88,6 @@ func shutdown(client MQTT.Client) {
 	}
 	wg.Wait()
 
-	client.Disconnect(20 * 1000) //wait 10sek at most
+	//we have to disconnect at last because one closeable unsubscripe all topics
+	client.Disconnect(10 * 1000) //wait 10sek at most
 }

internal/cmd/executor.go

@@ -27,6 +27,7 @@ func (c *CommandExecutor) ExecuteCommand(cmd string, args []string) ([]byte, err
 }
 
 func (c *CommandExecutor) ExecuteCommandWithContext(cmd string, args []string, executionContext context.Context) ([]byte, error) {
+	//register the context so that we have a chance to cancel the commands later
 	ctx := c.registerContext(executionContext)
 	c.openExecutions.Add(1)
 	defer c.openExecutions.Done()
@@ -46,16 +47,19 @@ func (c *CommandExecutor) ExecuteCommandWithContext(cmd string, args []string, e
 }
 
 func (c *CommandExecutor) registerContext(parentContext context.Context) context.Context {
+	//lock to ensure the map is thread-safe
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
+	//wrap the given context so that we can later cancel the context (see Close func)
 	ctx, cancelFunc := context.WithCancel(parentContext)
 	c.usedContext[ctx] = cancelFunc
 
 	return ctx
 }
 
 func (c *CommandExecutor) releaseContext(ctx context.Context) {
+	//lock to ensure the map is thread-safe
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
@@ -65,19 +69,23 @@ func (c *CommandExecutor) releaseContext(ctx context.Context) {
 func (c *CommandExecutor) Close(timeout time.Duration) error {
 	wg := sync.WaitGroup{}
 
+	//this lock ensures that no new commands can start
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
+	//cancel all in parallel
 	for ctx, cancelFunc := range c.usedContext {
 		wg.Add(1)
 
+		//trigger cancel-func and wait for it
 		go func(c context.Context, cf context.CancelFunc) {
 			defer wg.Done()
 			cf()         //call cancel
 			<-ctx.Done() //wait for cancellation
 		}(ctx, cancelFunc)
 	}
 
+	//wait for all commands to be stopped
 	wgChan := make(chan bool)
 	go func() {
 		wg.Wait()

internal/mqtt/config/topic.go

@@ -26,6 +26,7 @@ type Availability struct {
 type Trigger struct {
 	Name    string  `json:"name"`
 	Topic   string  `json:"topic"`
+	Icon    string  `json:"icon"`
 	Command Command `json:"command"`
 }
 
@@ -34,6 +35,7 @@ type Sensor struct {
 	ResultTopic string   `json:"topic"`
 	Interval    Interval `json:"interval"`
 	Unit        string   `json:"unit"`
+	Icon        string   `json:"icon"`
 	Command     Command  `json:"command"`
 }
 
@@ -42,15 +44,15 @@ type Command struct {
 	Arguments []string `json:"arguments"`
 }
 
-func LoadTopicConfiguration(configFile, deviceId string) (TopicConfigurations, error) {
-	file, err := os.Open(configFile)
+func LoadTopicConfiguration(configFilePath, deviceId string) (TopicConfigurations, error) {
+	configFile, err := os.Open(configFilePath)
 	if err != nil {
 		return TopicConfigurations{}, fmt.Errorf("error while opening topic configuration file: %w", err)
 	}
-	defer file.Close()
+	defer configFile.Close()
 
 	var topicConfig TopicConfigurations
-	err = json.NewDecoder(file).Decode(&topicConfig)
+	err = json.NewDecoder(configFile).Decode(&topicConfig)
 	if err != nil {
 		return TopicConfigurations{}, fmt.Errorf("could not read topic configuration file: %w", err)
 	}

internal/mqtt/hassio/client.go

@@ -13,20 +13,20 @@ import (
 )
 
 type generalConfig struct {
-	Name                string  `json:"name"`
-	AvailabilityTopic   string  `json:"avty_t,omitempty"`
-	PayloadAvailable    string  `json:"pl_avail,omitempty"`
-	PayloadNotAvailable string  `json:"pl_not_avail,omitempty"`
-	UniqueId            string  `json:"uniq_id"`
-	Device              *device `json:"dev,omitempty"`
+	Name                string `json:"name"`
+	AvailabilityTopic   string `json:"avty_t,omitempty"`
+	PayloadAvailable    string `json:"pl_avail,omitempty"`
+	PayloadNotAvailable string `json:"pl_not_avail,omitempty"`
+	UniqueId            string `json:"uniq_id"`
+	Icon                string `json:"ic,omitempty"`
+	Device              device `json:"dev,omitempty"`
 }
 
 type sensorConfig struct {
 	generalConfig
 
 	StateTopic      string `json:"stat_t"`
 	MeasurementUnit string `json:"unit_of_meas,omitempty"`
-	Icon            string `json:"ic,omitempty"`
 	ForceUpdate     *bool  `json:"frc_upd,omitempty"`
 }
 
@@ -82,7 +82,17 @@ func (c *Client) PublishDiscoveryConfig(config config.TopicConfigurations) {
 	//trigger
 	for _, trigger := range config.Trigger {
 		targetTopic := fmt.Sprintf("%sswitch/%s/%s/config", c.TopicPrefix, c.DeviceId, friendlyName(trigger.Name))
-		payload := c.generatePayloadForTriggerAction(config.Availability, trigger)
+		payload := c.generateSwitchPayloadForTriggerAction(config.Availability, trigger)
+		c.MqttClient.Publish(targetTopic, byte(0), false, payload)
+
+		//publish the trigger-result as sensor data
+		targetTopic = fmt.Sprintf("%ssensor/%s_%s/result/config", c.TopicPrefix, c.DeviceId, friendlyName(trigger.Name))
+		payload = c.generateResultPayloadForTriggerAction(config.Availability, trigger)
+		c.MqttClient.Publish(targetTopic, byte(0), false, payload)
+
+		//publish the trigger-state as sensor data
+		targetTopic = fmt.Sprintf("%ssensor/%s_%s/state/config", c.TopicPrefix, c.DeviceId, friendlyName(trigger.Name))
+		payload = c.generateStatePayloadForTriggerAction(config.Availability, trigger)
 		c.MqttClient.Publish(targetTopic, byte(0), false, payload)
 	}
 }
@@ -91,20 +101,24 @@ func friendlyName(name string) string {
 	return strings.Replace(name, " ", "_", -1)
 }
 
+func (c *Client) buildDevice() device {
+	return device{
+		Name:         c.DeviceName,
+		Ids:          []string{c.DeviceId},
+		Manufacturer: "rainu",
+		Model:        runtime.GOOS,
+		Version:      "mqtt-executor",
+	}
+}
+
 func (c *Client) generatePayloadForStatus(availability *config.Availability) []byte {
 	conf := sensorConfig{
 		generalConfig: generalConfig{
 			Name:                "Status",
 			PayloadAvailable:    availability.Payload.Available,
 			PayloadNotAvailable: availability.Payload.Unavailable,
 			UniqueId:            fmt.Sprintf("%s_status", c.DeviceId),
-			Device: &device{
-				Name:         c.DeviceName,
-				Ids:          []string{c.DeviceId},
-				Manufacturer: "rainu",
-				Model:        runtime.GOOS,
-				Version:      "mqtt-executor",
-			},
+			Device:              c.buildDevice(),
 		},
 		StateTopic: availability.Topic,
 	}
@@ -115,6 +129,7 @@ func (c *Client) generatePayloadForStatus(availability *config.Availability) []b
 
 	payload, err := json.Marshal(conf)
 	if err != nil {
+		//the "marshalling" is relatively safe - it should never appear at runtime
 		panic(err)
 	}
 	return payload
@@ -125,37 +140,31 @@ func (c *Client) generatePayloadForSensor(availability *config.Availability, sen
 	conf := sensorConfig{
 		generalConfig: generalConfig{
 			Name:     sensor.Name,
+			Icon:     sensor.Icon,
 			UniqueId: fmt.Sprintf("%s_%s", c.DeviceId, friendlyName(sensor.Name)),
-			Device: &device{
-				Ids: []string{c.DeviceId},
-			},
+			Device:   c.buildDevice(),
 		},
 		StateTopic:      sensor.ResultTopic,
 		MeasurementUnit: sensor.Unit,
-		Icon:            "",
 		ForceUpdate:     &bTrue,
 	}
-	if availability != nil {
-		conf.AvailabilityTopic = availability.Topic
-		conf.PayloadAvailable = availability.Payload.Available
-		conf.PayloadNotAvailable = availability.Payload.Unavailable
-	}
+	addAvailability(&conf.generalConfig, availability)
 
 	payload, err := json.Marshal(conf)
 	if err != nil {
+		//the "marshalling" is relatively safe - it should never appear at runtime
 		panic(err)
 	}
 	return payload
 }
 
-func (c *Client) generatePayloadForTriggerAction(availability *config.Availability, trigger config.Trigger) []byte {
+func (c *Client) generateSwitchPayloadForTriggerAction(availability *config.Availability, trigger config.Trigger) []byte {
 	conf := triggerConfig{
 		generalConfig: generalConfig{
 			Name:     fmt.Sprintf("%s", trigger.Name),
+			Icon:     trigger.Icon,
 			UniqueId: fmt.Sprintf("%s_%s", c.DeviceId, friendlyName(trigger.Name)),
-			Device: &device{
-				Ids: []string{c.DeviceId},
-			},
+			Device:   c.buildDevice(),
 		},
 		CommandTopic: trigger.Topic,
 		PayloadStart: mqtt.PayloadStart,
@@ -164,15 +173,60 @@ func (c *Client) generatePayloadForTriggerAction(availability *config.Availabili
 		StateRunning: mqtt.PayloadStatusRunning,
 		StateStopped: mqtt.PayloadStatusStopped,
 	}
-	if availability != nil {
-		conf.AvailabilityTopic = availability.Topic
-		conf.PayloadAvailable = availability.Payload.Available
-		conf.PayloadNotAvailable = availability.Payload.Unavailable
+	addAvailability(&conf.generalConfig, availability)
+
+	payload, err := json.Marshal(conf)
+	if err != nil {
+		//the "marshalling" is relatively safe - it should never appear at runtime
+		panic(err)
 	}
+	return payload
+}
+
+func (c *Client) generateResultPayloadForTriggerAction(availability *config.Availability, trigger config.Trigger) []byte {
+	conf := sensorConfig{
+		generalConfig: generalConfig{
+			Name:     fmt.Sprintf("%s - Result", trigger.Name),
+			Icon:     trigger.Icon,
+			UniqueId: fmt.Sprintf("%s_%s_result", c.DeviceId, friendlyName(trigger.Name)),
+			Device:   c.buildDevice(),
+		},
+		StateTopic: fmt.Sprintf("%s/%s", trigger.Topic, mqtt.TopicSuffixResult),
+	}
+	addAvailability(&conf.generalConfig, availability)
+
+	payload, err := json.Marshal(conf)
+	if err != nil {
+		//the "marshalling" is relatively safe - it should never appear at runtime
+		panic(err)
+	}
+	return payload
+}
+
+func (c *Client) generateStatePayloadForTriggerAction(availability *config.Availability, trigger config.Trigger) []byte {
+	conf := sensorConfig{
+		generalConfig: generalConfig{
+			Name:     fmt.Sprintf("%s - State", trigger.Name),
+			Icon:     trigger.Icon,
+			UniqueId: fmt.Sprintf("%s_%s_state", c.DeviceId, friendlyName(trigger.Name)),
+			Device:   c.buildDevice(),
+		},
+		StateTopic: fmt.Sprintf("%s/%s", trigger.Topic, mqtt.TopicSuffixState),
+	}
+	addAvailability(&conf.generalConfig, availability)
 
 	payload, err := json.Marshal(conf)
 	if err != nil {
+		//the "marshalling" is relatively safe - it should never appear at runtime
 		panic(err)
 	}
 	return payload
 }
+
+func addAvailability(config *generalConfig, availability *config.Availability) {
+	if availability != nil {
+		config.AvailabilityTopic = availability.Topic
+		config.PayloadAvailable = availability.Payload.Available
+		config.PayloadNotAvailable = availability.Payload.Unavailable
+	}
+}

internal/mqtt/sensor.go

@@ -20,6 +20,7 @@ type SensorWorker struct {
 
 func (s *SensorWorker) Initialise(publishQOS byte, sensorConfigs []config.Sensor) {
 
+	//generate a context so that we can cancel it later (see Close func)
 	var ctx context.Context
 	ctx, s.cancelFunc = context.WithCancel(context.Background())
 
@@ -32,24 +33,25 @@ func (s *SensorWorker) Initialise(publishQOS byte, sensorConfigs []config.Sensor
 func (s *SensorWorker) runSensor(ctx context.Context, publishQOS byte, sensorConf config.Sensor) {
 	defer s.waitGroup.Done()
 
-	//first one
-	s.executeCommand(publishQOS, sensorConf)
+	//first execution
+	s.executeCommand(ctx, publishQOS, sensorConf)
 
 	ticker := time.Tick(time.Duration(sensorConf.Interval))
 	for {
+		//wait until next tick or shutdown
 		select {
 		case <-ticker:
-			s.executeCommand(publishQOS, sensorConf)
+			s.executeCommand(ctx, publishQOS, sensorConf)
 		case <-ctx.Done():
 			return
 		}
 	}
 }
 
-func (s *SensorWorker) executeCommand(publishQOS byte, sensorConf config.Sensor) {
-	output, execErr := s.Executor.ExecuteCommand(sensorConf.Command.Name, sensorConf.Command.Arguments)
+func (s *SensorWorker) executeCommand(ctx context.Context, publishQOS byte, sensorConf config.Sensor) {
+	output, execErr := s.Executor.ExecuteCommandWithContext(sensorConf.Command.Name, sensorConf.Command.Arguments, ctx)
 	if execErr != nil {
-		s.MqttClient.Publish(sensorConf.ResultTopic, publishQOS, false, "<FAILED> "+execErr.Error())
+		s.MqttClient.Publish(sensorConf.ResultTopic, publishQOS, false, "<FAILED>;"+execErr.Error())
 		return
 	}
 
@@ -58,6 +60,7 @@ func (s *SensorWorker) executeCommand(publishQOS byte, sensorConf config.Sensor)
 
 func (s *SensorWorker) Close(timeout time.Duration) error {
 	if s.cancelFunc != nil {
+		//close the context to interrupt possible running commands
 		s.cancelFunc()
 	}