From 832459ee7ccccb7c8ad3dfb8f58258a20061d9a9 Mon Sep 17 00:00:00 2001 From: Cyrille Nofficial Date: Sat, 30 Nov 2019 19:03:07 +0100 Subject: [PATCH] Initial commit --- .gitignore | 1 + arduino/arduino_test.go | 148 ++++++++++++++++++++++++++++++++++ mode/mode.go | 0 mqttdevice/mqttdevice.go | 85 +++++++++++++++++++ mqttdevice/mqttdevice_test.go | 51 ++++++++++++ testtools/testtools.go | 37 +++++++++ 6 files changed, 322 insertions(+) create mode 100644 .gitignore create mode 100644 arduino/arduino_test.go create mode 100644 mode/mode.go create mode 100644 mqttdevice/mqttdevice.go create mode 100644 mqttdevice/mqttdevice_test.go create mode 100644 testtools/testtools.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bfa6a22 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +# Created by .ignore support plugin (hsz.mobi) diff --git a/arduino/arduino_test.go b/arduino/arduino_test.go new file mode 100644 index 0000000..aba97bf --- /dev/null +++ b/arduino/arduino_test.go @@ -0,0 +1,148 @@ +package arduino + +import ( + "bufio" + "fmt" + "net" + "robocar/mode" + "testing" + "time" +) + +func TestArduinoPart_Update(t *testing.T) { + ln, err := net.Listen("tcp", ":8080") + if err != nil { + t.Fatalf("unable to init connection for test") + } + defer ln.Close() + + client, err := net.Dial("tcp", "localhost:8080") + if err != nil { + t.Fatalf("unable to init connection for test") + } + defer client.Close() + + conn, err := ln.Accept() + if err != nil { + t.Fatalf("unable to init connection for test") + } + defer conn.Close() + + a := ArduinoPart{serial: conn} + go a.Run() + + channel1, channel2, channel3, channel4, channel5, channel6, distanceCm := 678, 910, 1112, 1678, 1910, 112, 128 + cases := []struct { + name, content string + expectedThrottle, expectedSteering float32 + expectedDriveMode mode.DriveMode + expectedSwitchRecord bool + expectedDistanceCm int + }{ + {"Good value", + fmt.Sprintf("12345,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, channel5, channel6, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"Unparsable line", + "12350,invalid line\n", + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"Switch record on", + fmt.Sprintf("12355,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, 998, channel6, distanceCm), + -1., -1., mode.DriveModeUser, true, distanceCm}, + + {"Switch record off", + fmt.Sprintf("12360,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, 1987, channel6, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"Switch record off", + fmt.Sprintf("12365,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, 1850, channel6, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"Switch record on", + fmt.Sprintf("12370,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, 1003, channel6, distanceCm), + -1., -1., mode.DriveModeUser, true, distanceCm}, + + + {"DriveMode: user", + fmt.Sprintf("12375,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, channel5, 998, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"DriveMode: pilot", + fmt.Sprintf("12380,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, channel5, 1987, distanceCm), + -1., -1., mode.DriveModePilot, false, distanceCm}, + {"DriveMode: pilot", + fmt.Sprintf("12385,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, channel5, 1850, distanceCm), + -1., -1., mode.DriveModePilot, false, distanceCm}, + + // DriveMode: user + {"DriveMode: user", + fmt.Sprintf("12390,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, channel5, 1003, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + + + {"Sterring: over left", + fmt.Sprintf("12395,%d,%d,%d,%d,%d,%d,50,%d\n", 99, channel2, channel3, channel4, channel5, channel6, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"Sterring: left", + fmt.Sprintf("12400,%d,%d,%d,%d,%d,%d,50,%d\n", 998, channel2, channel3, channel4, channel5, channel6, distanceCm), + -1., -0.93, mode.DriveModeUser, false, distanceCm}, + {"Sterring: middle", + fmt.Sprintf("12405,%d,%d,%d,%d,%d,%d,50,%d\n", 1450, channel2, channel3, channel4, channel5, channel6, distanceCm), + -1., -0.04, mode.DriveModeUser, false, distanceCm}, + {"Sterring: right", + fmt.Sprintf("12410,%d,%d,%d,%d,%d,%d,50,%d\n", 1958, channel2, channel3, channel4, channel5, channel6, distanceCm), + -1., 0.96, mode.DriveModeUser, false, distanceCm}, + {"Sterring: over right", + fmt.Sprintf("12415,%d,%d,%d,%d,%d,%d,50,%d\n", 2998, channel2, channel3, channel4, channel5, channel6, distanceCm), + -1., 1., mode.DriveModeUser, false, distanceCm}, + + + {"Throttle: over down", + fmt.Sprintf("12420,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, 99, channel3, channel4, channel5, channel6, distanceCm), + -1., -1., mode.DriveModeUser, false, distanceCm}, + {"Throttle: down", + fmt.Sprintf("12425,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, 998, channel3, channel4, channel5, channel6, distanceCm), + -0.95, -1., mode.DriveModeUser, false, distanceCm}, + {"Throttle: stop", + fmt.Sprintf("12430,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, 1450, channel3, channel4, channel5, channel6, distanceCm), + -0.03, -1., mode.DriveModeUser, false, distanceCm}, + {"Throttle: up", + fmt.Sprintf("12435,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, 1948, channel3, channel4, channel5, channel6, distanceCm), + 0.99, -1., mode.DriveModeUser, false, distanceCm}, + {"Throttle: over up", + fmt.Sprintf("12440,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, 2998, channel3, channel4, channel5, channel6, distanceCm), + 1., -1., mode.DriveModeUser, false, distanceCm}, + + {"Distance cm", + fmt.Sprintf("12445,%d,%d,%d,%d,%d,%d,50,%d\n", channel1, channel2, channel3, channel4, channel5, channel6, 43), + -1., -1., mode.DriveModeUser, false, 43}, + } + + for _, c := range cases { + w := bufio.NewWriter(client) + _, err := w.WriteString(c.content) + if err != nil { + t.Errorf("unable to send test content: %v", c.content) + } + err = w.Flush() + if err != nil { + t.Error("unable to flush content") + } + + time.Sleep(1* time.Millisecond) + a.mutex.Lock() + if fmt.Sprintf("%0.2f", a.throttle) != fmt.Sprintf("%0.2f", c.expectedThrottle) { + t.Errorf("%s: bad throttle value, expected: %0.2f, actual: %.2f", c.name, c.expectedThrottle, a.throttle) + } + if fmt.Sprintf("%0.2f", a.steering) != fmt.Sprintf("%0.2f", c.expectedSteering) { + t.Errorf("%s: bad steering value, expected: %0.2f, actual: %.2f", c.name, c.expectedSteering, a.steering) + } + if a.driveMode != c.expectedDriveMode { + t.Errorf("%s: bad drive mode, expected: %v, actual:%v", c.name, c.expectedDriveMode, a.driveMode) + } + if a.ctrlRecord != c.expectedSwitchRecord { + t.Errorf("%s: bad switch record, expected: %v, actual:%v", c.name, c.expectedSwitchRecord, a.ctrlRecord) + } + if a.distanceCm != c.expectedDistanceCm { + t.Errorf("%s: bad distanceCm, expected: %v" + + ", actual:%v", c.name, c.expectedDistanceCm, a.distanceCm) + } + a.mutex.Unlock() + } +} diff --git a/mode/mode.go b/mode/mode.go new file mode 100644 index 0000000..e69de29 diff --git a/mqttdevice/mqttdevice.go b/mqttdevice/mqttdevice.go new file mode 100644 index 0000000..44eb232 --- /dev/null +++ b/mqttdevice/mqttdevice.go @@ -0,0 +1,85 @@ +package mqttdevice + +import ( + "fmt" + MQTT "github.com/eclipse/paho.mqtt.golang" + "io" + "log" +) + +type Publisher interface { + Publish(topic string, payload interface{}) +} + +type Subscriber interface { + Subscribe(topic string, mh MQTT.MessageHandler) +} + +type MQTTPubSub interface { + Publisher + Subscriber + io.Closer +} + +type pahoMqttPubSub struct { + Uri string + Username string + Password string + ClientId string + Qos int + Retain bool + client MQTT.Client +} + +func NewPahoMqttPubSub(uri string, username string, password string, clientId string, qos int, retain bool) MQTTPubSub { + p := pahoMqttPubSub{Uri: uri, Username: username, Password: password, ClientId: clientId, Qos: qos, Retain: retain} + p.Connect() + return &p +} + +// Publish message to broker +func (p *pahoMqttPubSub) Publish(topic string, payload interface{}) { + tokenResp := p.client.Publish(topic, byte(p.Qos), p.Retain, payload) + if tokenResp.Error() != nil { + log.Fatalf("%+v\n", tokenResp.Error()) + } +} + +// Register func to execute on message +func (p *pahoMqttPubSub) Subscribe(topic string, callback MQTT.MessageHandler) { + tokenResp := p.client.Subscribe(topic, byte(p.Qos), callback) + if tokenResp.Error() != nil { + log.Fatalf("%+v\n", tokenResp.Error()) + } +} + +// Close connection to broker +func (p *pahoMqttPubSub) Close() error { + p.client.Disconnect(500) + return nil +} + +func (p *pahoMqttPubSub) Connect() { + if p.client != nil && p.client.IsConnected() { + return + } + //create a ClientOptions struct setting the broker address, clientid, turn + //off trace output and set the default message handler + opts := MQTT.NewClientOptions().AddBroker(p.Uri) + opts.SetUsername(p.Username) + opts.SetPassword(p.Password) + opts.SetClientID(p.ClientId) + opts.SetAutoReconnect(true) + opts.SetDefaultPublishHandler( + //define a function for the default message handler + func(client MQTT.Client, msg MQTT.Message) { + fmt.Printf("TOPIC: %s\n", msg.Topic()) + fmt.Printf("MSG: %s\n", msg.Payload()) + }) + + //create and start a client using the above ClientOptions + p.client = MQTT.NewClient(opts) + if token := p.client.Connect(); token.Wait() && token.Error() != nil { + panic(token.Error()) + } +} diff --git a/mqttdevice/mqttdevice_test.go b/mqttdevice/mqttdevice_test.go new file mode 100644 index 0000000..94c22f6 --- /dev/null +++ b/mqttdevice/mqttdevice_test.go @@ -0,0 +1,51 @@ +package mqttdevice + +import ( + mqtt "github.com/eclipse/paho.mqtt.golang" + "testing" + "warmup4ie2mqtt/testtools" +) + +func TestIntegration(t *testing.T) { + + ctx, mqttC, mqttUri := testtools.MqttContainer(t) + defer mqttC.Terminate(ctx) + + t.Run("ConnectAndClose", func(t *testing.T) { + t.Logf("Mqtt connection %s ready", mqttUri) + + p := pahoMqttPubSub{Uri: mqttUri, ClientId: "TestMqtt", Username: "guest", Password: "guest"} + p.Connect() + p.Close() + }) + t.Run("Publish", func(t *testing.T) { + options := mqtt.NewClientOptions().AddBroker(mqttUri) + options.SetUsername("guest") + options.SetPassword("guest") + + client := mqtt.NewClient(options) + token := client.Connect() + defer client.Disconnect(100) + token.Wait() + if token.Error() != nil { + t.Fatalf("unable to connect to mqtt broker: %v\n", token.Error()) + } + + c := make(chan string) + defer close(c) + client.Subscribe("test/publish", 0, func(client mqtt.Client, message mqtt.Message) { + c <- string(message.Payload()) + }).Wait() + + p := pahoMqttPubSub{Uri: mqttUri, ClientId: "TestMqtt", Username: "guest", Password: "guest"} + p.Connect() + defer p.Close() + + p.Publish("test/publish", "Test1234") + result := <-c + if result != "Test1234" { + t.Fatalf("bad message: %v\n", result) + } + + }) +} diff --git a/testtools/testtools.go b/testtools/testtools.go new file mode 100644 index 0000000..85c3056 --- /dev/null +++ b/testtools/testtools.go @@ -0,0 +1,37 @@ +package testtools + +import ( + "context" + "fmt" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" + "testing" +) + +func MqttContainer(t *testing.T) (context.Context, testcontainers.Container, string) { + ctx := context.Background() + req := testcontainers.ContainerRequest{ + Image: "eclipse-mosquitto", + ExposedPorts: []string{"1883/tcp"}, + WaitingFor: wait.ForLog("listen socket on port 1883."), + } + mqttC, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + if err != nil { + t.Error(err) + } + + ip, err := mqttC.Host(ctx) + if err != nil { + t.Error(err) + } + port, err := mqttC.MappedPort(ctx, "1883/tcp") + if err != nil { + t.Error(err) + } + + mqttUri := fmt.Sprintf("tcp://%s:%d", ip, port.Int()) + return ctx, mqttC, mqttUri +}