Base project

internal/csv/helper.go

@@ -0,0 +1,54 @@
+package csv
+
+import (
+	"encoding/csv"
+	"os"
+
+	"go.uber.org/fx"
+	"go.uber.org/zap"
+)
+
+type Helper struct {
+	fx.In
+	csvFile   *os.File
+	csvWriter *csv.Writer
+	logger    *zap.Logger
+}
+
+func NewHelper(lc fx.Lifecycle, logger *zap.Logger) *Helper {
+	csvFile, err := os.OpenFile("mqtt-logs.csv", os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
+	if err != nil {
+		logger.Fatal("error opening csv file", zap.Error(err))
+	}
+
+	csvWriter := csv.NewWriter(csvFile)
+
+	return &Helper{
+		csvFile:   csvFile,
+		csvWriter: csvWriter,
+		logger:    logger,
+	}
+}
+
+func (h *Helper) AddEntry(date string, topic string, message string) error {
+	record := []string{date, topic, message}
+	if err := h.csvWriter.Write(record); err != nil {
+		h.logger.Error("error writing to csv file", zap.Error(err))
+		return err
+	}
+
+	h.csvWriter.Flush()
+
+	return nil
+}
+
+func (h *Helper) Close() error {
+	h.csvWriter.Flush()
+
+	if err := h.csvFile.Close(); err != nil {
+		h.logger.Error("error closing csv file", zap.Error(err))
+		return err
+	}
+
+	return nil
+}

internal/csv/init.go

@@ -0,0 +1,33 @@
+package csv
+
+import (
+	"encoding/csv"
+	"io"
+	"os"
+
+	"go.uber.org/zap"
+)
+
+func InitializeHeader(logger *zap.Logger) {
+	csvFile, err := os.OpenFile("mqtt-logs.csv", os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
+	if err != nil {
+		logger.Fatal("error opening csv file", zap.Error(err))
+	}
+	defer csvFile.Close()
+
+	r := csv.NewReader(csvFile)
+	w := csv.NewWriter(csvFile)
+	record, err := r.Read()
+
+	if err == io.EOF {
+		w.Write([]string{"Date (UTC)", "Topic", "Message"})
+		w.Flush()
+	} else if err != nil {
+		logger.Fatal("error initializing csv file", zap.Error(err))
+	} else if record[0] == "Date (UTC)" && record[1] == "Topic" && record[2] == "Message" {
+		return
+	} else {
+		logger.Fatal("First row of csv file is not as expected. Please make sure the file is in the correct format.")
+	}
+
+}

internal/mqtt/logger.go

@@ -0,0 +1,24 @@
+package mqtt
+
+import (
+	"log"
+	"strings"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+)
+
+type ZapWriter struct {
+    sugar *zap.SugaredLogger
+}
+
+func (z *ZapWriter) Write(p []byte) (n int, err error) {
+    z.sugar.Debug(strings.TrimSpace(string(p)))
+    return len(p), nil
+}
+
+func SetupMQTTLogging(logger *zap.Logger) {
+    sugar := logger.Sugar()
+    mqtt.DEBUG = log.New(&ZapWriter{sugar: sugar}, "[DEBUG] ", 0)
+    mqtt.ERROR = log.New(&ZapWriter{sugar: sugar}, "[ERROR] ", 0)
+}

internal/mqtt/mqtt.go

@@ -0,0 +1,52 @@
+package mqtt
+
+import (
+	"MQTTLogger/config"
+	"MQTTLogger/internal/csv"
+	"time"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+)
+
+func NewClient(logger *zap.Logger, config *config.Config, csvHelper *csv.Helper) mqtt.Client {
+	var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
+		csvHelper.AddEntry(time.Now().UTC().Format("2006-01-02 15:04:05"), msg.Topic(), string(msg.Payload()))
+	}
+
+	opts := mqtt.NewClientOptions().AddBroker(config.URI).SetClientID("mqtt-logger")
+	opts.SetKeepAlive(2 * time.Second)
+	opts.SetDefaultPublishHandler(f)
+	opts.SetPingTimeout(1 * time.Second)
+	opts.SetConnectionNotificationHandler(func(client mqtt.Client, notification mqtt.ConnectionNotification) {
+		switch n := notification.(type) {
+		case mqtt.ConnectionNotificationConnected:
+			logger.Info("Connected to MQTT broker")
+		case mqtt.ConnectionNotificationConnecting:
+			logger.Info("connecting", zap.Bool("isReconnect", n.IsReconnect), zap.Int("attempt", n.Attempt))
+		case mqtt.ConnectionNotificationFailed:
+			logger.Error("connection failed", zap.String("error", n.Reason.Error()))
+		case mqtt.ConnectionNotificationLost:
+			logger.Error("connection lost", zap.String("error", n.Reason.Error()))
+		case mqtt.ConnectionNotificationBroker:
+			logger.Info("broker", zap.String("name", n.Broker.String()))
+		case mqtt.ConnectionNotificationBrokerFailed:
+			logger.Error("broker failed", zap.String("broker", n.Broker.String()), zap.String("error", n.Reason.Error()))
+		}
+	})
+
+	c := mqtt.NewClient(opts)
+	if token := c.Connect(); token.Wait() && token.Error() != nil {
+		logger.Fatal("error connecting to MQTT broker", zap.Error(token.Error()))
+	}
+
+	return c
+}
+
+func StartListening(logger *zap.Logger, config *config.Config, client mqtt.Client) {
+	for _, topic := range config.Topics {
+		if token := client.Subscribe(topic, 0, nil); token.Wait() && token.Error() != nil {
+			logger.Error("error subscribing to topic", zap.String("topic", topic), zap.Error(token.Error()))
+		}
+	}
+}