Skip to content

Commit eabedb6

Browse files
rainurainu
committed
implements mechanism to interrupt running trigger-commands
1 parent 2bffef9 commit eabedb6

File tree

4 files changed

+108
-36
lines changed

4 files changed

+108
-36
lines changed

‎internal/cmd/executor.go

+4-1
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,11 @@ func (c *CommandExecutor) ExecuteCommandWithContext(cmd string, args []string, e
3535
command := exec.CommandContext(ctx, cmd, args...)
3636
out, execErr := command.CombinedOutput()
3737

38-
if execErr != nil {
38+
if execErr != nil && ctx.Err() == nil {
3939
zap.L().Error("Command execution failed.", zap.Error(execErr))
40+
} else if ctx.Err() != nil {
41+
zap.L().Info("Command execution cancelled.")
42+
return out, ctx.Err()
4043
}
4144

4245
return out, execErr

‎internal/mqtt/config/topic.go

+3-5
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,8 @@ type Availability struct {
2424
}
2525

2626
type Trigger struct {
27-
Name string `json:"name"`
28-
Topic string `json:"topic"`
29-
Actions TriggerAction `json:"actions"`
30-
}
31-
type TriggerAction map[string]struct {
27+
Name string `json:"name"`
28+
Topic string `json:"topic"`
3229
Command Command `json:"command"`
3330
}
3431

@@ -91,5 +88,6 @@ func (t *TopicConfigurations) validate() error {
9188
//TODO
9289
//keine wildcards in Topic-Namen
9390
//Keine Sonderzeichen in Payloads
91+
//Trigger-Name muss einzigartig sein
9492
return nil
9593
}

‎internal/mqtt/hassio/client.go

+8-10
Original file line numberDiff line numberDiff line change
@@ -81,11 +81,9 @@ func (c *Client) PublishDiscoveryConfig(config config.TopicConfigurations) {
8181

8282
//trigger
8383
for _, trigger := range config.Trigger {
84-
for action := range trigger.Actions {
85-
targetTopic := fmt.Sprintf("%sswitch/%s_%s/%s/config", c.TopicPrefix, c.DeviceId, friendlyName(trigger.Name), friendlyName(action))
86-
payload := c.generatePayloadForTriggerAction(config.Availability, trigger, action)
87-
c.MqttClient.Publish(targetTopic, byte(0), false, payload)
88-
}
84+
targetTopic := fmt.Sprintf("%sswitch/%s/%s/config", c.TopicPrefix, c.DeviceId, friendlyName(trigger.Name))
85+
payload := c.generatePayloadForTriggerAction(config.Availability, trigger)
86+
c.MqttClient.Publish(targetTopic, byte(0), false, payload)
8987
}
9088
}
9189

@@ -150,19 +148,19 @@ func (c *Client) generatePayloadForSensor(availability *config.Availability, sen
150148
return payload
151149
}
152150

153-
func (c *Client) generatePayloadForTriggerAction(availability *config.Availability, trigger config.Trigger, action string) []byte {
151+
func (c *Client) generatePayloadForTriggerAction(availability *config.Availability, trigger config.Trigger) []byte {
154152
conf := triggerConfig{
155153
generalConfig: generalConfig{
156-
Name: fmt.Sprintf("%s - %s", trigger.Name, action),
157-
UniqueId: fmt.Sprintf("%s_%s_%s", c.DeviceId, friendlyName(trigger.Name), friendlyName(action)),
154+
Name: fmt.Sprintf("%s", trigger.Name),
155+
UniqueId: fmt.Sprintf("%s_%s", c.DeviceId, friendlyName(trigger.Name)),
158156
Device: &device{
159157
Ids: []string{c.DeviceId},
160158
},
161159
},
162160
CommandTopic: trigger.Topic,
163-
PayloadStart: action,
161+
PayloadStart: mqtt.PayloadStart,
164162
PayloadStop: mqtt.PayloadStop,
165-
StateTopic: fmt.Sprintf("%s/%s/%s", trigger.Topic, action, mqtt.TopicSuffixState),
163+
StateTopic: fmt.Sprintf("%s/%s", trigger.Topic, mqtt.TopicSuffixState),
166164
StateRunning: mqtt.PayloadStatusRunning,
167165
StateStopped: mqtt.PayloadStatusStopped,
168166
}

‎internal/mqtt/trigger.go

+93-20
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@ import (
77
"github.com/rainu/mqtt-executor/internal/cmd"
88
"github.com/rainu/mqtt-executor/internal/mqtt/config"
99
"go.uber.org/zap"
10+
"strings"
11+
"sync"
1012
"time"
1113
)
1214

@@ -15,58 +17,129 @@ const (
1517
TopicSuffixResult = "RESULT"
1618
PayloadStatusRunning = "RUNNING"
1719
PayloadStatusStopped = "STOPPED"
20+
PayloadStart = "START"
1821
PayloadStop = "STOP"
1922
)
2023

2124
type Trigger struct {
22-
triggerConfigs []config.Trigger
25+
lock sync.RWMutex
26+
runningCommands map[string]context.CancelFunc
27+
triggerConfigs []config.Trigger
28+
publishQOS byte
2329

2430
Executor *cmd.CommandExecutor
2531
MqttClient MQTT.Client
2632
}
2733

2834
func (t *Trigger) Initialise(subscribeQOS, publishQOS byte, triggerConfigs []config.Trigger) {
35+
t.publishQOS = publishQOS
36+
t.runningCommands = map[string]context.CancelFunc{}
2937

3038
for _, triggerConf := range triggerConfigs {
3139
t.triggerConfigs = append(t.triggerConfigs, triggerConf)
32-
t.MqttClient.Subscribe(triggerConf.Topic, subscribeQOS, t.createTriggerHandler(publishQOS, triggerConf))
40+
t.MqttClient.Subscribe(triggerConf.Topic, subscribeQOS, t.createTriggerHandler(triggerConf))
41+
t.publishStatus(triggerConf.Topic, PayloadStatusStopped)
3342
}
34-
35-
//TODO: publish stopped state for each command (inital state)
3643
}
3744

38-
func (t *Trigger) createTriggerHandler(publishQOS byte, triggerConfig config.Trigger) MQTT.MessageHandler {
45+
func (t *Trigger) createTriggerHandler(triggerConfig config.Trigger) MQTT.MessageHandler {
3946
return func(client MQTT.Client, message MQTT.Message) {
4047
zap.L().Info("Incoming message: ",
4148
zap.String("topic", message.Topic()),
4249
zap.ByteString("payload", message.Payload()),
4350
)
4451

45-
action, exists := triggerConfig.Actions[string(message.Payload())]
46-
if !exists {
47-
zap.L().Warn("Command is not configured")
48-
return
52+
action := strings.ToUpper(string(message.Payload()))
53+
54+
switch action {
55+
case PayloadStart:
56+
if t.isCommandRunning(triggerConfig) {
57+
zap.L().Warn("Command is already running. Skip execution!", zap.String("trigger", triggerConfig.Name))
58+
return
59+
}
60+
61+
go t.executeCommand(message.Topic(), triggerConfig)
62+
case PayloadStop:
63+
if !t.isCommandRunning(triggerConfig) {
64+
return
65+
}
66+
t.interruptCommand(triggerConfig)
67+
t.unregisterCommand(triggerConfig)
68+
default:
69+
zap.L().Warn("Invalid payload")
4970
}
50-
cmd := action.Command
51-
52-
go t.executeCommand(publishQOS, message.Topic(), string(message.Payload()), cmd)
5371
}
5472
}
5573

56-
func (t *Trigger) executeCommand(publishQOS byte, topic, action string, command config.Command) {
57-
stateTopic := fmt.Sprintf("%s/%s/%s", topic, action, TopicSuffixState)
58-
resultTopic := fmt.Sprintf("%s/%s/%s", topic, action, TopicSuffixResult)
74+
func (t *Trigger) isCommandRunning(trigger config.Trigger) bool {
75+
t.lock.RLock()
76+
defer t.lock.RUnlock()
77+
78+
_, exist := t.runningCommands[trigger.Name]
79+
return exist
80+
}
81+
82+
func (t *Trigger) registerCommand(trigger config.Trigger) context.Context {
83+
t.lock.Lock()
84+
defer t.lock.Unlock()
85+
86+
ctx, cancelFunc := context.WithCancel(context.Background())
87+
t.runningCommands[trigger.Name] = cancelFunc
88+
89+
return ctx
90+
}
91+
92+
func (t *Trigger) unregisterCommand(trigger config.Trigger) {
93+
t.lock.Lock()
94+
defer t.lock.Unlock()
95+
96+
delete(t.runningCommands, trigger.Name)
97+
}
98+
99+
func (t *Trigger) interruptCommand(trigger config.Trigger) {
100+
t.lock.RLock()
101+
defer t.lock.RUnlock()
59102

60-
t.MqttClient.Publish(stateTopic, publishQOS, false, PayloadStatusRunning)
61-
defer t.MqttClient.Publish(stateTopic, publishQOS, false, PayloadStatusStopped)
103+
//execute corresponding cancel func
104+
t.runningCommands[trigger.Name]()
105+
}
106+
107+
func (t *Trigger) executeCommand(topic string, trigger config.Trigger) {
108+
ctx := t.registerCommand(trigger)
109+
defer t.unregisterCommand(trigger)
110+
111+
t.publishStatus(topic, PayloadStatusRunning)
112+
defer t.publishStatus(topic, PayloadStatusStopped)
62113

63-
output, execErr := t.Executor.ExecuteCommandWithContext(command.Name, command.Arguments, context.Background())
114+
output, execErr := t.Executor.ExecuteCommandWithContext(trigger.Command.Name, trigger.Command.Arguments, ctx)
64115
if execErr != nil {
65-
t.MqttClient.Publish(resultTopic, publishQOS, false, "<FAILED> "+execErr.Error())
116+
if execErr == context.Canceled {
117+
t.publishResult(topic, "<INTERRUPTED>")
118+
} else {
119+
t.publishResult(topic, "<FAILED>;"+execErr.Error())
120+
}
66121
return
67122
}
68123

69-
t.MqttClient.Publish(resultTopic, publishQOS, false, output)
124+
t.publishResult(topic, output)
125+
}
126+
127+
func (t *Trigger) publishStatus(parentTopic, status string) MQTT.Token {
128+
stateTopic := t.buildStateTopic(parentTopic)
129+
return t.MqttClient.Publish(stateTopic, t.publishQOS, false, status)
130+
}
131+
132+
func (t *Trigger) publishResult(parentTopic string, result interface{}) MQTT.Token {
133+
resultTopic := t.buildResultTopic(parentTopic)
134+
return t.MqttClient.Publish(resultTopic, t.publishQOS, false, result)
135+
}
136+
137+
func (t *Trigger) buildStateTopic(parentTopic string) string {
138+
return fmt.Sprintf("%s/%s", parentTopic, TopicSuffixState)
139+
}
140+
141+
func (t *Trigger) buildResultTopic(parentTopic string) string {
142+
return fmt.Sprintf("%s/%s", parentTopic, TopicSuffixResult)
70143
}
71144

72145
func (t *Trigger) Close(timeout time.Duration) error {

0 commit comments

Comments
 (0)