Develop (#4)

* first working version

* updating license

* better msg handler

* adding some tests

* trigger travis

* update glide

* travis

* some changes

* travis

* update all

* trying stuff for travis

* update go in travis

* Update .travis.yml

OSX test never ending

* workig config parser + tests

* Revert "workig config parser + tests"

This reverts commit 1100621.

* working config parser + test

* update readme

* Format/gofmt (#3)

* changes for gofmt and golint

* Update README.md

* Update README.md

* removing debug msg

README.md

@@ -1,9 +1,11 @@
 # Mqttbeat
 [![Build Status](https://travis-ci.org/nathan-K-/mqttbeat.svg?branch=master)](https://travis-ci.org/nathan-K-/mqttbeat)
+[![Go Report Card](https://goreportcard.com/badge/github.com/nathan-K-/mqttbeat)](https://goreportcard.com/report/github.com/nathan-K-/mqttbeat)
 
 Welcome to Mqttbeat.
 
 This beat will allow you to put MQTT messages in an elasticsearch instance.
+You can use [this little project](https://github.com/nathan-K-/stack-docker) to easily start an elastic stack.
 
 Ensure that this folder is at the following location:
 `${GOPATH}/github.com/nathan-k-/mqttbeat`

beater/mqttbeat.go

@@ -1,11 +1,11 @@
 package beater
 
 import (
-	"fmt"
-	"time"
 	"encoding/json"
-	"strings"
+	"fmt"
 	"strconv"
+	"strings"
+	"time"
 
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
@@ -19,31 +19,32 @@ import (
 	"github.com/nathan-k-/mqttbeat/config"
 )
 
+// Mqttbeat represent a mqtt beat object
 type Mqttbeat struct {
-	done   chan struct{}
-	beatConfig config.Config
+	done          chan struct{}
+	beatConfig    config.Config
 	elasticClient publisher.Client
-	mqttClient MQTT.Client
+	mqttClient    MQTT.Client
 }
 
 // Prepare mqtt client
 func setupMqttClient(bt *Mqttbeat) {
 	mqttClientOpt := MQTT.NewClientOptions()
-	mqttClientOpt.AddBroker(bt.beatConfig.BrokerUrl)
-	logp.Info("BROKER url " + bt.beatConfig.BrokerUrl)
+	mqttClientOpt.AddBroker(bt.beatConfig.BrokerURL)
+	logp.Info("BROKER url " + bt.beatConfig.BrokerURL)
 
 	bt.mqttClient = MQTT.NewClient(mqttClientOpt)
 }
 
-// Creates beater
+// New function creates our mqtt beater
 func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 	config := config.DefaultConfig
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, fmt.Errorf("Error reading config file: %v", err)
 	}
 
 	bt := &Mqttbeat{
-		done:   make(chan struct{}),
+		done:       make(chan struct{}),
 		beatConfig: config,
 	}
 	setupMqttClient(bt)
@@ -54,18 +55,16 @@ func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 	logp.Info("MQTT Client connected")
 
 	subscriptions := ParseTopics(bt.beatConfig.TopicsSubscribe)
-		//bt.beatConfig.TopicsSubscribe
+	//bt.beatConfig.TopicsSubscribe
 
 	// Mqtt client - Subscribe to every topic in the config file, and bind with message handler
-	if token := bt.mqttClient.SubscribeMultiple(subscriptions, bt.onMessage);
-	token.Wait() && token.Error() != nil {
+	if token := bt.mqttClient.SubscribeMultiple(subscriptions, bt.onMessage); token.Wait() && token.Error() != nil {
 		panic(token.Error())
 	}
 
 	return bt, nil
 }
 
-
 // Mqtt message handler
 func (bt *Mqttbeat) onMessage(client MQTT.Client, msg MQTT.Message) {
 	logp.Info("MQTT MESSAGE RECEIVED " + string(msg.Payload()))
@@ -79,15 +78,15 @@ func (bt *Mqttbeat) onMessage(client MQTT.Client, msg MQTT.Message) {
 		event["payload"] = msg.Payload()
 	}
 
-	event["beat"]= common.MapStr{"index": "mqttbeat", "type":"message"}
+	event["beat"] = common.MapStr{"index": "mqttbeat", "type": "message"}
 	event["@timestamp"] = common.Time(time.Now())
 	event["topic"] = msg.Topic()
 	// Finally sending the message to elasticsearch
 	bt.elasticClient.PublishEvent(event)
 	logp.Info("Event sent")
-	}
-
+}
 
+// Run is used to start this beater, once configured and connected
 func (bt *Mqttbeat) Run(b *beat.Beat) error {
 	logp.Info("mqttbeat is running! Hit CTRL-C to stop it.")
 	bt.elasticClient = b.Publisher.Connect()
@@ -101,48 +100,50 @@ func (bt *Mqttbeat) Run(b *beat.Beat) error {
 	}
 }
 
+// Stop is used to close this beater
 func (bt *Mqttbeat) Stop() {
 	bt.mqttClient.Disconnect(250)
 	bt.elasticClient.Close()
 	close(bt.done)
 }
 
-// Helpers
+// DecodePayload will try to decode the payload. If every check fails, it will
+// return the payload as a string
 func DecodePayload(payload []byte) common.MapStr {
 	event := make(common.MapStr)
 
 	// A msgpack payload must be a json-like object
 	err := msgpack.Unmarshal(payload, &event)
-	if  err == nil {
+	if err == nil {
 		logp.Info("Payload decoded - msgpack")
 		return event
 	}
 
 	err = json.Unmarshal(payload, &event)
-	if  err == nil {
+	if err == nil {
 		logp.Info("Payload decoded - json")
 		return event
 	}
 
 	// default case
-	event["payload"]= string(payload)
+	event["payload"] = string(payload)
 	logp.Info("Payload decoded - text")
 	return event
 }
 
-func ParseTopics(topics []string) map[string]byte{
+// ParseTopics will parse the config file and return a map with topic:QoS
+func ParseTopics(topics []string) map[string]byte {
 	subscriptions := make(map[string]byte)
-	for _, value := range topics{
+	for _, value := range topics {
 		// Fist, spliting the string topic?qos
 		topic, qosStr := strings.Split(value, "?")[0], strings.Split(value, "?")[1]
 		// Then, parsing the qos to an int
 		qosInt, err := strconv.ParseInt(qosStr, 10, 0)
 		if err != nil {
 			panic("Error parsing topics")
 		}
-		// Finally, filling the subscriptions map 
+		// Finally, filling the subscriptions map
 		subscriptions[topic] = byte(qosInt)
 	}
-	fmt.Println(subscriptions)
 	return subscriptions
-}
+}

beater/mqttbeat_test.go

@@ -3,12 +3,11 @@ package beater
 import (
 	"github.com/elastic/beats/libbeat/common"
 	"gopkg.in/vmihailenco/msgpack.v2"
-	"testing"
 	"reflect"
+	"testing"
 )
 
-
-func TestDecodeMsgpackJson(t *testing.T){
+func TestDecodeMsgpackJson(t *testing.T) {
 
 	reference := make(common.MapStr)
 	reference["hello"] = "world"
@@ -22,7 +21,7 @@ func TestDecodeMsgpackJson(t *testing.T){
 	}
 }
 
-func TestDecodeJson(t *testing.T){
+func TestDecodeJson(t *testing.T) {
 
 	reference := make(common.MapStr)
 	reference["hello"] = "world"
@@ -35,7 +34,7 @@ func TestDecodeJson(t *testing.T){
 	}
 }
 
-func TestDecodeText(t *testing.T){
+func TestDecodeText(t *testing.T) {
 	payload := "Bonjour, monde!"
 
 	reference := make(common.MapStr)
@@ -48,7 +47,7 @@ func TestDecodeText(t *testing.T){
 	}
 }
 
-func TestParseTopic(t *testing.T){
+func TestParseTopic(t *testing.T) {
 	input := []string{"some/topic?0", "some/ohter/topic?2", "final/topic?1"}
 
 	reference := map[string]byte{"some/topic": 0, "some/ohter/topic": 2, "final/topic": 1}
@@ -58,4 +57,4 @@ func TestParseTopic(t *testing.T){
 	if !reflect.DeepEqual(reference, output) {
 		t.Error("Not equals")
 	}
-}
+}

config/config.go

@@ -5,15 +5,17 @@ package config
 
 import "time"
 
+// Config represents every needed configuration fields
 type Config struct {
-	BrokerUrl string `config:"broker_url"`
-	TopicsSubscribe []string `config:"topics_subscribe"`
-	DecodePaylod bool `config:"decode_payload"`
+	BrokerURL       string        `config:"broker_url"`
+	TopicsSubscribe []string      `config:"topics_subscribe"`
+	DecodePaylod    bool          `config:"decode_payload"`
 	Period          time.Duration `config:"period"`
 }
 
+// DefaultConfig will be used if no config file is founded
 var DefaultConfig = Config{
-	BrokerUrl: "tcp://localhost:1883",
+	BrokerURL:       "tcp://localhost:1883",
 	TopicsSubscribe: []string{"/test/mqttbeat/#?1"},
-	DecodePaylod: true,
+	DecodePaylod:    true,
 }