Initial commit

Author: rainu

.gitignore

@@ -0,0 +1,17 @@
+.idea
+
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2020 rainu
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,39 @@
+# mqtt-executor
+
+A simple MQTT client written in go that subscribes to a configurable list of MQTT topics on the specified broker and 
+executes a given shell script/command whenever a message arrives.
+
+# Configuration
+Create a configuration file named "config.json"
+```json
+{
+  "sys/file": {
+    "create": ["/usr/bin/touch", "/tmp/example"],
+    "remove": ["/bin/rm", "-f", "/tmp/example"],
+    "info": ["/bin/ls", "-l", "/tmp/example"]
+  }
+}
+```
+
+# Usage
+
+Start the tool with the path to the config file and the URL of the MQTT broker
+```bash
+mqtt-executor -broker tcp://127.0.0.1:1883 -config /path/to/config.json
+```
+
+# Trigger command execution
+
+```bash
+mosquitto_pub -d -t sys/file -m "create"
+mosquitto_pub -d -t sys/file -m "info"
+mosquitto_pub -d -t sys/file -m "remove"
+```
+
+# Get the command results
+
+Each command result will be written in **<incomingTopicName>/RESULT**. For example:
+
+```bash
+mosquitto_sub -d -t sys/file/RESULT
+```

cmd/mqtt-executor/cmd_executor.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	MQTT "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+	"os/exec"
+	"sync"
+	"time"
+)
+
+type CommandExecutor struct {
+	lock           sync.RWMutex
+	usedContext    map[context.Context]context.CancelFunc
+	openExecutions sync.WaitGroup
+}
+
+var commandExecutor CommandExecutor
+
+func init() {
+	commandExecutor = CommandExecutor{
+		lock:        sync.RWMutex{},
+		usedContext: map[context.Context]context.CancelFunc{},
+	}
+}
+
+func (c *CommandExecutor) ExecuteCommand(client MQTT.Client, message MQTT.Message, commandAndArgs []string) {
+	ctx := c.createContext()
+	c.openExecutions.Add(1)
+	defer c.openExecutions.Done()
+
+	command := exec.CommandContext(ctx, commandAndArgs[0], commandAndArgs[1:]...)
+	out, execErr := command.CombinedOutput()
+	if execErr != nil {
+		zap.L().Error("Command execution failed.", zap.Error(execErr))
+		client.Publish(fmt.Sprintf("%s/RESULT", message.Topic()), byte(*Config.PublishQOS), false, "<FAILED> "+execErr.Error())
+		return
+	}
+
+	client.Publish(fmt.Sprintf("%s/RESULT", message.Topic()), byte(*Config.PublishQOS), false, out)
+	c.releaseContext(ctx)
+}
+
+func (c *CommandExecutor) createContext() context.Context {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	c.usedContext[ctx] = cancelFunc
+
+	return ctx
+}
+
+func (c *CommandExecutor) releaseContext(ctx context.Context) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	delete(c.usedContext, ctx)
+}
+
+func (c *CommandExecutor) Close(timeout time.Duration) error {
+	wg := sync.WaitGroup{}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	for ctx, cancelFunc := range c.usedContext {
+		wg.Add(1)
+
+		go func(c context.Context, cf context.CancelFunc) {
+			defer wg.Done()
+			cf()         //call cancel
+			<-ctx.Done() //wait for cancellation
+		}(ctx, cancelFunc)
+	}
+
+	wgChan := make(chan bool)
+	go func() {
+		wg.Wait()
+		c.openExecutions.Wait()
+
+		wgChan <- true
+	}()
+
+	//wait for WaitGroup or Timeout
+	select {
+	case <-wgChan:
+		return nil
+	case <-time.After(timeout):
+		return errors.New("timeout exceeded")
+	}
+}

cmd/mqtt-executor/config.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	MQTT "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+	"os"
+)
+
+type config struct {
+	Broker       *string
+	SubscribeQOS *int
+	PublishQOS   *int
+	Username     *string
+	Password     *string
+	ClientId     *string
+
+	TopicConfigFile     *string
+	TopicConfigurations TopicConfigurations
+}
+
+var Config config
+
+type TopicConfigurations map[string]TopicConfiguration
+type TopicConfiguration map[string][]string
+
+func LoadConfig() {
+	Config = config{
+		Broker:       flag.String("broker", "", "The broker URI. ex: tcp://127.0.0.1:1883"),
+		SubscribeQOS: flag.Int("sub-qos", 0, "The Quality of Service for subscription 0,1,2 (default 0)"),
+		PublishQOS:   flag.Int("pub-qos", 0, "The Quality of Service for publishing 0,1,2 (default 0)"),
+		Username:     flag.String("user", "", "The User (optional)"),
+		Password:     flag.String("password", "", "The password (optional)"),
+		ClientId:     flag.String("client-id", "mqtt-executor", "The ClientID (optional)"),
+
+		TopicConfigFile: flag.String("config", "./config.json", "The topic configuration file"),
+	}
+	flag.Parse()
+
+	if *Config.Broker == "" {
+		zap.L().Fatal("Broker is missing!")
+	}
+	if *Config.SubscribeQOS != 0 && *Config.SubscribeQOS != 1 && *Config.SubscribeQOS != 2 {
+		zap.L().Fatal("Invalid qos level!")
+	}
+	if *Config.PublishQOS != 0 && *Config.PublishQOS != 1 && *Config.PublishQOS != 2 {
+		zap.L().Fatal("Invalid qos level!")
+	}
+	if *Config.TopicConfigFile == "" {
+		zap.L().Fatal("Topic configuration file is missing!")
+	} else {
+		file, err := os.Open(*Config.TopicConfigFile)
+		if err != nil {
+			zap.L().Fatal("Error while opening topic configuration file: %s", zap.Error(err))
+		}
+		defer file.Close()
+
+		err = json.NewDecoder(file).Decode(&Config.TopicConfigurations)
+		if err != nil {
+			zap.L().Fatal("Could not read topic configuration file: %s", zap.Error(err))
+		}
+	}
+}
+
+func (c *config) GetMQTTOpts() *MQTT.ClientOptions {
+	opts := MQTT.NewClientOptions()
+
+	opts.AddBroker(*c.Broker)
+	if *c.Username != "" {
+		opts.SetUsername(*c.Username)
+	}
+	if *c.Password != "" {
+		opts.SetPassword(*c.Password)
+	}
+	if *c.ClientId != "" {
+		opts.SetClientID(*c.ClientId)
+	}
+
+	return opts
+}

cmd/mqtt-executor/log.go

@@ -0,0 +1,9 @@
+package main
+
+import "go.uber.org/zap"
+
+func init() {
+	logger, _ := zap.NewDevelopment()
+	zap.ReplaceGlobals(logger)
+	defer zap.L().Sync()
+}

cmd/mqtt-executor/main.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	MQTT "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+)
+
+func main() {
+	LoadConfig()
+
+	signals := make(chan os.Signal, 1)
+	defer close(signals)
+	signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
+
+	client := MQTT.NewClient(Config.GetMQTTOpts())
+
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		zap.L().Fatal("Error while connecting to mqtt broker: %s", zap.Error(token.Error()))
+	}
+
+	for topicName, config := range Config.TopicConfigurations {
+		client.Subscribe(topicName, byte(*Config.SubscribeQOS), createMessageHandler(config))
+	}
+
+	// wait for interrupt
+	<-signals
+
+	shutdown(client)
+}
+
+func createMessageHandler(topicConfig TopicConfiguration) MQTT.MessageHandler {
+	return func(client MQTT.Client, message MQTT.Message) {
+		zap.L().Info("Incoming message: ",
+			zap.String("topic", message.Topic()),
+			zap.ByteString("payload", message.Payload()),
+		)
+
+		cmdArgs, exists := topicConfig[string(message.Payload())]
+		if !exists {
+			zap.L().Warn("Command is not configured")
+			return
+		}
+
+		go commandExecutor.ExecuteCommand(client, message, cmdArgs)
+	}
+}
+
+func shutdown(client MQTT.Client) {
+	zap.L().Info("Shutting down...")
+	for topicName, _ := range Config.TopicConfigurations {
+		client.Unsubscribe(topicName)
+	}
+
+	err := commandExecutor.Close(20 * time.Second)
+	if err != nil {
+		zap.L().Error("Timeout while waiting command execution!", zap.Error(err))
+	}
+
+	client.Disconnect(20 * 1000) //wait 10sek at most
+}

go.mod

@@ -0,0 +1,9 @@
+module github.com/rainu/mqtt-executor
+
+go 1.14
+
+require (
+	github.com/eclipse/paho.mqtt.golang v1.2.0
+	go.uber.org/zap v1.15.0
+	golang.org/x/net v0.0.0-20200501053045-e0ff5e5a1de5 // indirect
+)