refactor: improve MQTT client implementation and error handling

Author: pmoscode

mqtt/client.go

@@ -1,8 +1,8 @@
 package mqtt
 
 import (
+	"context"
 	"log"
-	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/pmoscode/go-common/shutdown"
@@ -13,7 +13,7 @@ const notConnected = "Mqtt Client not connected! Call 'connect' method..."
 
 // Client wraps the paho MQTT client with convenience methods for connect, publish, subscribe, and disconnect.
 type Client struct {
-	client  *mqtt.Client
+	client  mqtt.Client
 	options *mqtt.ClientOptions
 }
 
@@ -34,15 +34,15 @@ func (c *Client) connect(client mqtt.Client) error {
 		return token.Error()
 	}
 
-	c.client = &client
+	c.client = client
 	log.Println("Mqtt connected to", c.options.Servers[0])
 
 	return nil
 }
 
 // Disconnect does a clean disconnect from the mqtt broker.
 func (c *Client) Disconnect() error {
-	(*c.client).Disconnect(100)
+	c.client.Disconnect(100)
 
 	return nil
 }
@@ -53,7 +53,7 @@ func (c *Client) Publish(message *Message) {
 	if c.client == nil {
 		log.Println(notConnected)
 	} else {
-		token := (*c.client).Publish(message.Topic, 2, false, message.FromJson())
+		token := c.client.Publish(message.Topic, 2, false, message.FromJson())
 		token.Wait()
 	}
 }
@@ -64,7 +64,7 @@ func (c *Client) Subscribe(topic string, fn func(message Message)) {
 	if c.client == nil {
 		log.Println(notConnected)
 	} else {
-		(*c.client).Subscribe(topic, 2, func(client mqtt.Client, msg mqtt.Message) {
+		c.client.Subscribe(topic, 2, func(client mqtt.Client, msg mqtt.Message) {
 			message := Message{
 				Topic: msg.Topic(),
 				Value: msg.Payload(),
@@ -74,15 +74,13 @@ func (c *Client) Subscribe(topic string, fn func(message Message)) {
 	}
 }
 
-// LoopForever Halts the current thread.
-func (c *Client) LoopForever() {
+// LoopForever halts the current thread until the context is cancelled.
+// On cancellation, the client will be disconnected cleanly.
+func (c *Client) LoopForever(ctx context.Context) {
 	if c.client == nil {
 		log.Println(notConnected)
 	} else {
 		shutdown.GetObserver().AddCommand(c.Disconnect)
-
-		for {
-			time.Sleep(10 * time.Second)
-		}
+		<-ctx.Done()
 	}
 }

mqtt/message-type.go

@@ -3,7 +3,6 @@ package mqtt
 import (
 	"encoding/json"
 	"fmt"
-	"log"
 )
 
 // Message defines the default message which is received and sent by the mqtt client.
@@ -24,16 +23,24 @@ func (m Message) FromJson() string {
 	return string(marshal)
 }
 
-// ToStruct converts the received JSON message to the given target struct
-func (m Message) ToStruct(target any) {
-	message := m.ToRawString()
-	err := json.Unmarshal([]byte(message), &target)
+// ToStruct converts the received JSON message to the given target struct.
+// Returns an error if the message value is not a byte slice or not valid JSON.
+func (m Message) ToStruct(target any) error {
+	message, err := m.ToRawString()
 	if err != nil {
-		log.Println("Message is not a valid Json: ", message)
+		return err
 	}
+	if err := json.Unmarshal([]byte(message), &target); err != nil {
+		return fmt.Errorf("message is not valid JSON: %w", err)
+	}
+	return nil
 }
 
-// ToRawString converts the received JSON string to an ordinary string
-func (m Message) ToRawString() string {
-	return string(m.Value.([]uint8))
+// ToRawString converts the received message value to an ordinary string.
+// Returns an error if the value is not a byte slice.
+func (m Message) ToRawString() (string, error) {
+	if v, ok := m.Value.([]uint8); ok {
+		return string(v), nil
+	}
+	return "", fmt.Errorf("message value is not a byte slice, got %T", m.Value)
 }

mqtt/mqtt_test.go

@@ -5,6 +5,8 @@ import (
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestConnection(t *testing.T) {
@@ -15,9 +17,7 @@ func TestConnection(t *testing.T) {
 		WithUsernameAndPassword("test", "pwd")
 
 	brokerOpt, err := hostConfig.Build()
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	client := NewClient(
 		WithClientId("test"),
@@ -26,21 +26,13 @@ func TestConnection(t *testing.T) {
 
 	mockClientImpl := &mockClient{}
 	err = client.connect(mockClientImpl)
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	err = client.Disconnect()
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
-	if mockClientImpl.cntConnect != 1 {
-		t.Fatal("Wrong connect count: ", mockClientImpl.cntConnect, " should be ", 1)
-	}
-	if mockClientImpl.cntDisconnect != 1 {
-		t.Fatal("Wrong disconnect count: ", mockClientImpl.cntDisconnect, " should be ", 1)
-	}
+	assert.Equal(t, 1, mockClientImpl.cntConnect)
+	assert.Equal(t, 1, mockClientImpl.cntDisconnect)
 }
 
 func TestPublish(t *testing.T) {
@@ -50,9 +42,7 @@ func TestPublish(t *testing.T) {
 		WithProtocol(MqttTcp)
 
 	brokerOpt, err := hostConfig.Build()
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	client := NewClient(
 		WithClientId("test"),
@@ -61,29 +51,19 @@ func TestPublish(t *testing.T) {
 
 	mockClientImpl := &mockClient{}
 	err = client.connect(mockClientImpl)
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
 
 	client.Publish(&Message{
 		Topic: "/test/testMessage",
 		Value: "{'test': 2}",
 	})
 
 	err = client.Disconnect()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	if mockClientImpl.cntConnect != 1 {
-		t.Fatal("Wrong connect count: ", mockClientImpl.cntConnect, " should be ", 1)
-	}
-	if mockClientImpl.cntDisconnect != 1 {
-		t.Fatal("Wrong disconnect count: ", mockClientImpl.cntDisconnect, " should be ", 1)
-	}
-	if mockClientImpl.cntPublish != 1 {
-		t.Fatal("Wrong publish count: ", mockClientImpl.cntPublish, " should be ", 1)
-	}
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, mockClientImpl.cntConnect)
+	assert.Equal(t, 1, mockClientImpl.cntDisconnect)
+	assert.Equal(t, 1, mockClientImpl.cntPublish)
 }
 
 type mockClient struct {