From f2ab59f4d0919e663432cc4c560ada37fdf05331 Mon Sep 17 00:00:00 2001 From: Cyrille Nofficial Date: Mon, 27 Jan 2020 19:21:26 +0100 Subject: [PATCH] [cli] Split topic prefix arg --- arduino/arduino.go | 39 +++++++++++++++++++----------------- arduino/arduino_test.go | 11 +++++++++- cmd/rc-arduino/rc-arduino.go | 12 +++++++---- 3 files changed, 39 insertions(+), 23 deletions(-) diff --git a/arduino/arduino.go b/arduino/arduino.go index 17e1988..0761007 100644 --- a/arduino/arduino.go +++ b/arduino/arduino.go @@ -29,7 +29,7 @@ var ( type Part struct { client mqtt.Client - topicBase string + throttleTopic, steeringTopic, driveModeTopic, switchRecordTopic string pubFrequency float64 serial io.Reader mutex sync.Mutex @@ -41,7 +41,7 @@ type Part struct { cancel chan interface{} } -func NewPart(client mqtt.Client, name string, baud int, topicBase string, pubFrequency float64, debug bool) *Part { +func NewPart(client mqtt.Client, name string, baud int, throttleTopic, steeringTopic, driveModeTopic, switchRecordTopic string, pubFrequency float64, debug bool) *Part { c := &serial.Config{Name: name, Baud: baud} s, err := serial.OpenPort(c) if err != nil { @@ -50,7 +50,10 @@ func NewPart(client mqtt.Client, name string, baud int, topicBase string, pubFre return &Part{ client: client, serial: s, - topicBase: topicBase, + throttleTopic: throttleTopic, + steeringTopic: steeringTopic, + driveModeTopic:driveModeTopic, + switchRecordTopic:switchRecordTopic, pubFrequency: pubFrequency, driveMode: events.DriveMode_INVALID, debug: debug, @@ -192,13 +195,12 @@ func (a *Part) processChannel6(v string) { } func (a *Part) publishLoop() { - prefix := strings.TrimSuffix(a.topicBase, "/") ticker := time.NewTicker(time.Second / time.Duration(int(a.pubFrequency))) for { select { case <-ticker.C: - a.publishValues(prefix) + a.publishValues() case <-a.cancel: ticker.Stop() return @@ -206,17 +208,17 @@ func (a *Part) publishLoop() { } } -func (a *Part) publishValues(prefix string) { +func (a *Part) publishValues() { a.mutex.Lock() defer a.mutex.Unlock() - a.publishThrottle(prefix) - a.publishSteering(prefix) - a.publishDriveMode(prefix) - a.publishSwitchRecord(prefix) + a.publishThrottle() + a.publishSteering() + a.publishDriveMode() + a.publishSwitchRecord() } -func (a *Part) publishThrottle(prefix string) { +func (a *Part) publishThrottle() { throttle := events.ThrottleMessage{ Throttle: a.throttle, Confidence: 1.0, @@ -226,10 +228,11 @@ func (a *Part) publishThrottle(prefix string) { log.Errorf("unable to marshal protobuf throttle message: %v", err) return } - publish(a.client, prefix+"/throttle", &throttleMessage) + log.Infof("throttle channel: %v", a.throttle) + publish(a.client, a.throttleTopic, &throttleMessage) } -func (a *Part) publishSteering(prefix string) { +func (a *Part) publishSteering() { steering := events.SteeringMessage{ Steering: a.steering, Confidence: 1.0, @@ -239,10 +242,10 @@ func (a *Part) publishSteering(prefix string) { log.Errorf("unable to marshal protobuf steering message: %v", err) return } - publish(a.client, prefix+"/steering", &steeringMessage) + publish(a.client, a.steeringTopic, &steeringMessage) } -func (a *Part) publishDriveMode(prefix string) { +func (a *Part) publishDriveMode() { dm := events.DriveModeMessage{ DriveMode: a.driveMode, } @@ -251,10 +254,10 @@ func (a *Part) publishDriveMode(prefix string) { log.Errorf("unable to marshal protobuf driveMode message: %v", err) return } - publish(a.client, prefix+"/drive_mode", &driveModeMessage) + publish(a.client, a.driveModeTopic, &driveModeMessage) } -func (a *Part) publishSwitchRecord(prefix string) { +func (a *Part) publishSwitchRecord() { sr := events.SwitchRecordMessage{ Enabled: a.ctrlRecord, } @@ -263,7 +266,7 @@ func (a *Part) publishSwitchRecord(prefix string) { log.Errorf("unable to marshal protobuf SwitchRecord message: %v", err) return } - publish(a.client, prefix+"/switch_record", &switchRecordMessage) + publish(a.client, a.switchRecordTopic, &switchRecordMessage) } var publish = func(client mqtt.Client, topic string, payload *[]byte) { diff --git a/arduino/arduino_test.go b/arduino/arduino_test.go index bebe507..8d57e8a 100644 --- a/arduino/arduino_test.go +++ b/arduino/arduino_test.go @@ -192,7 +192,16 @@ func TestPublish(t *testing.T) { defer conn.Close() pubFrequency := 100. - a := Part{client: nil, serial: conn, pubFrequency: pubFrequency, topicBase: "car/part/arduino/", cancel:make(chan interface{})} + a := Part{ + client: nil, + serial: conn, + pubFrequency: pubFrequency, + throttleTopic: "car/part/arduino/throttle", + steeringTopic: "car/part/arduino/steering", + driveModeTopic: "car/part/arduino/drive_mode", + switchRecordTopic: "car/part/arduino/switch_record", + cancel: make(chan interface{}), + } go a.Start() defer a.Stop() diff --git a/cmd/rc-arduino/rc-arduino.go b/cmd/rc-arduino/rc-arduino.go index c4f0691..fba7d6e 100644 --- a/cmd/rc-arduino/rc-arduino.go +++ b/cmd/rc-arduino/rc-arduino.go @@ -11,7 +11,8 @@ import ( const DefaultClientId = "robocar-arduino" func main() { - var mqttBroker, username, password, clientId, topicBase string + var mqttBroker, username, password, clientId string + var throttleTopic, steeringTopic, driveModeTopic, switchRecordTopic string var device string var baud int var pubFrequency float64 @@ -23,7 +24,10 @@ func main() { cli.InitMqttFlags(DefaultClientId, &mqttBroker, &username, &password, &clientId, &mqttQos, &mqttRetain) flag.Float64Var(&pubFrequency, "mqtt-pub-frequency", 25., "Number of messages to publish per second") - flag.StringVar(&topicBase, "mqtt-topic-base", os.Getenv("MQTT_TOPIC_BASE"), "Mqtt topic prefix, use MQTT_TOPIC_BASE if args not set") + flag.StringVar(&throttleTopic, "mqtt-topic-throttle", os.Getenv("MQTT_TOPIC_THROTTLE"), "Mqtt topic where to publish throttle values, use MQTT_TOPIC_THROTTLE if args not set") + flag.StringVar(&steeringTopic, "mqtt-topic-steering", os.Getenv("MQTT_TOPIC_STEERING"), "Mqtt topic where to publish steering values, use MQTT_TOPIC_STEERING if args not set") + flag.StringVar(&driveModeTopic, "mqtt-topic-drive-mode", os.Getenv("MQTT_TOPIC_DRIVE_MODE"), "Mqtt topic where to publish drive mode state, use MQTT_TOPIC_DRIVE_MODE if args not set") + flag.StringVar(&switchRecordTopic, "mqtt-topic-switch-record", os.Getenv("MQTT_TOPIC_SWITCH_RECORD"), "Mqtt topic where to publish switch record state, use MQTT_TOPIC_SWITCH_RECORD if args not set") flag.StringVar(&device, "device", "/dev/serial0", "Serial device") flag.IntVar(&baud, "baud", 115200, "Serial baud") flag.BoolVar(&debug, "debug", false, "Display raw value to debug") @@ -35,12 +39,12 @@ func main() { } client, err := cli.Connect(mqttBroker, username, password, clientId) - if err != nil{ + if err != nil { log.Fatalf("unable to connect to mqtt broker: %v", err) } defer client.Disconnect(10) - a := arduino.NewPart(client, device, baud, topicBase, pubFrequency, debug) + a := arduino.NewPart(client, device, baud, throttleTopic, steeringTopic, driveModeTopic, switchRecordTopic, pubFrequency, debug) err = a.Start() if err != nil { log.Printf("unable to start service: %v", err)