[refactor] Read protobuf messages

vendor/github.com/cyrilix/robocar-base/cli/cli.go

@@ -2,7 +2,7 @@ package cli
 
 import (
 	"flag"
-	"fmt"
+	"github.com/cyrilix/robocar-base/mqttdevice"
 	"github.com/cyrilix/robocar-base/service"
 	MQTT "github.com/eclipse/paho.mqtt.golang"
 	"log"
@@ -92,24 +92,5 @@ func InitFloat64Flag(key string, defValue float64) float64 {
 }
 
 func Connect(uri, username, password, clientId string) (MQTT.Client, error) {
-	//create a ClientOptions struct setting the broker address, clientid, turn
-	//off trace output and set the default message handler
-	opts := MQTT.NewClientOptions().AddBroker(uri)
-	opts.SetUsername(username)
-	opts.SetPassword(password)
-	opts.SetClientID(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
-	client := MQTT.NewClient(opts)
-	if token := client.Connect(); token.Wait() && token.Error() != nil {
-		return nil, fmt.Errorf("unable to connect to mqtt bus: %v", token.Error())
-	}
-	return client, nil
+	return mqttdevice.Connect(uri, username, password, clientId)
 }

vendor/github.com/cyrilix/robocar-base/mqttdevice/mqttdevice.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"github.com/cyrilix/robocar-base/types"
 	MQTT "github.com/eclipse/paho.mqtt.golang"
-	"io"
 	"log"
 	"strconv"
 )
@@ -21,57 +20,26 @@ type Subscriber interface {
 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
+	qos      int
+	retain   bool
 }
 
-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()
+func NewPahoMqttPubSub(client MQTT.Client, qos int, retain bool) MQTTPubSub {
+	p := pahoMqttPubSub{client: client, qos: qos, retain: retain}
 	return &p
 }
 
-// Publish message to broker
-func (p *pahoMqttPubSub) Publish(topic string, payload MqttValue) {
-	tokenResp := p.client.Publish(topic, byte(p.Qos), p.Retain, string(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
-	}
+func Connect(uri, username, password, clientId string) (MQTT.Client, error) {
 	//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 := MQTT.NewClientOptions().AddBroker(uri)
+	opts.SetUsername(username)
+	opts.SetPassword(password)
+	opts.SetClientID(clientId)
 	opts.SetAutoReconnect(true)
 	opts.SetDefaultPublishHandler(
 		//define a function for the default message handler
@@ -81,9 +49,26 @@ func (p *pahoMqttPubSub) Connect() {
 		})
 
 	//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())
+	client := MQTT.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		return nil, fmt.Errorf("unable to connect to mqtt bus: %v", token.Error())
+	}
+	return client, nil
+}
+
+// Publish message to broker
+func (p *pahoMqttPubSub) Publish(topic string, payload MqttValue) {
+	tokenResp := p.client.Publish(topic, byte(p.qos), p.retain, string(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())
 	}
 }
 

vendor/github.com/cyrilix/robocar-base/testtools/testtools.go

@@ -3,6 +3,8 @@ package testtools
 import (
 	"github.com/cyrilix/robocar-base/mqttdevice"
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/golang/protobuf/proto"
+	log "github.com/sirupsen/logrus"
 	"sync"
 )
 
@@ -71,3 +73,12 @@ func NewFakeMessage(topic string, payload []byte) mqtt.Message {
 		acked:   false,
 	}
 }
+
+func NewFakeMessageFromProtobuf(topic string, msg proto.Message) mqtt.Message{
+	payload, err := proto.Marshal(msg)
+	if err  != nil {
+		log.Errorf("unable to marshal protobuf message %T: %v", msg, err)
+		return nil
+	}
+	return NewFakeMessage(topic, payload)
+}