Refactor MQTT handling into separate package

- Created new mqtt package with simple client interface - Implemented buffered channel for decoded messages - Updated main.go to use the new MQTT client - Added tests for the MQTT client - Simplified message handling to focus on 'e' and 'map' formats - Added TODO for handling JSON format messages 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-06 08:43:12 +02:00 · 2025-04-20 18:11:15 -07:00
parent 5bc956af55
commit 879877fa53
3 changed files with 244 additions and 83 deletions
@@ -7,11 +7,9 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
-	"time"

 	"meshstream/decoder"
-
-	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"meshstream/mqtt"
 )

 const (
@@ -21,52 +19,6 @@ const (
 	mqttTopicPrefix = "msh/US/bayarea"
 )

-var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
-	fmt.Printf("Received message from topic: %s\n", msg.Topic())
-
-	// Parse the topic structure
-	topicInfo, err := decoder.ParseTopic(msg.Topic())
-	if err != nil {
-		fmt.Printf("Error parsing topic: %v\n", err)
-		fmt.Printf("Raw topic: %s\n", msg.Topic())
-		fmt.Printf("Raw payload: %x\n", msg.Payload())
-	} else {
-		// First decode the message based on its format
-		var formattedOutput string
-		switch topicInfo.Format {
-		case "e", "c", "map":
-			// Binary encoded protobuf message (both regular and map formats use the same decoder)
-			decodedPacket := decoder.DecodeMessage(msg.Payload(), topicInfo)
-			formattedOutput = decoder.FormatTopicAndPacket(topicInfo, decodedPacket)
-		case "json":
-			// JSON format message
-			jsonData, err := decoder.DecodeJSONMessage(msg.Payload())
-			if err != nil {
-				fmt.Printf("Error decoding JSON message: %v\n", err)
-				formattedOutput = decoder.FormatTopicAndRawData(topicInfo, msg.Payload())
-			} else {
-				formattedOutput = decoder.FormatTopicAndJSONData(topicInfo, jsonData)
-			}
-		default:
-			// Unsupported format
-			formattedOutput = decoder.FormatTopicAndRawData(topicInfo, msg.Payload())
-		}
-
-		// Print the formatted output
-		fmt.Println(formattedOutput)
-	}
-
-	fmt.Println(strings.Repeat("-", 80))
-}
-
-var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
-	fmt.Println("Connected to MQTT Broker!")
-}
-
-var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
-	fmt.Printf("Connection lost: %v\n", err)
-}
-
 func main() {
 	// Set up logging
 	log.SetOutput(os.Stdout)
@@ -81,40 +33,46 @@ func main() {
 		log.Printf("Failed to initialize ERSN channel key: %v", err)
 	}

-	// Create MQTT client options
-	opts := mqtt.NewClientOptions()
-	opts.AddBroker(fmt.Sprintf("tcp://%s:1883", mqttBroker))
-	opts.SetClientID("meshstream-client")
-	opts.SetUsername(mqttUsername)
-	opts.SetPassword(mqttPassword)
-	opts.SetDefaultPublishHandler(messagePubHandler)
-	opts.SetPingTimeout(1 * time.Second)
-	opts.OnConnect = connectHandler
-	opts.OnConnectionLost = connectLostHandler
-
-	// Create and start a client
-	client := mqtt.NewClient(opts)
-	if token := client.Connect(); token.Wait() && token.Error() != nil {
-		log.Fatalf("Error connecting to MQTT broker: %v", token.Error())
+	// Configure and create the MQTT client
+	mqttConfig := mqtt.Config{
+		Broker:   mqttBroker,
+		Username: mqttUsername,
+		Password: mqttPassword,
+		ClientID: "meshstream-client",
+		Topic:    mqttTopicPrefix + "/#",
 	}
-
-	// Subscribe to all topics for this region
-	// This will capture:
-	// - msh/US/CA/Motherlode/2/e/# (binary protobuf data)
-	// - msh/US/CA/Motherlode/2/json/# (JSON formatted data)
-	topic := mqttTopicPrefix + "/#"
-	token := client.Subscribe(topic, 0, nil)
-	token.Wait()
-	fmt.Printf("Subscribed to topic: %s\n", topic)
-
-	// Wait for interrupt signal to gracefully shutdown
+	
+	mqttClient := mqtt.NewClient(mqttConfig)
+	
+	// Connect to the MQTT broker
+	if err := mqttClient.Connect(); err != nil {
+		log.Fatalf("Failed to connect to MQTT broker: %v", err)
+	}
+	
+	// Get the messages channel to receive decoded messages
+	messagesChan := mqttClient.Messages()
+	
+	// Setup signal handling for graceful shutdown
 	sig := make(chan os.Signal, 1)
 	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
-	<-sig
-
-	// Unsubscribe and disconnect
-	fmt.Println("Unsubscribing and disconnecting...")
-	token = client.Unsubscribe(topic)
-	token.Wait()
-	client.Disconnect(250)
-}
+	
+	// Process messages until interrupt received
+	fmt.Println("Waiting for messages... Press Ctrl+C to exit")
+	
+	// Main event loop
+	for {
+		select {
+		case msg := <-messagesChan:
+			// Format and print the decoded message
+			formattedOutput := decoder.FormatTopicAndPacket(msg.TopicInfo, msg.DecodedPacket)
+			fmt.Println(formattedOutput)
+			fmt.Println(strings.Repeat("-", 80))
+			
+		case <-sig:
+			// Got an interrupt signal, shutting down
+			fmt.Println("Shutting down...")
+			mqttClient.Disconnect()
+			return
+		}
+	}
+}
@@ -0,0 +1,134 @@
+package mqtt
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"meshstream/decoder"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+// Config holds configuration for the MQTT client
+type Config struct {
+	Broker   string
+	Username string
+	Password string
+	ClientID string
+	Topic    string
+}
+
+// Client manages the MQTT connection and message processing
+type Client struct {
+	config          Config
+	client          mqtt.Client
+	decodedMessages chan DecodedMessage
+	done            chan struct{}
+}
+
+// DecodedMessage contains a decoded packet and its topic info
+type DecodedMessage struct {
+	TopicInfo     *decoder.TopicInfo
+	DecodedPacket *decoder.DecodedPacket
+}
+
+// NewClient creates a new MQTT client with the provided configuration
+func NewClient(config Config) *Client {
+	return &Client{
+		config:          config,
+		decodedMessages: make(chan DecodedMessage, 100), // Buffer up to 100 messages
+		done:            make(chan struct{}),
+	}
+}
+
+// Connect establishes a connection to the MQTT broker
+func (c *Client) Connect() error {
+	// Create MQTT client options
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(fmt.Sprintf("tcp://%s:1883", c.config.Broker))
+	opts.SetClientID(c.config.ClientID)
+	opts.SetUsername(c.config.Username)
+	opts.SetPassword(c.config.Password)
+	opts.SetDefaultPublishHandler(c.messageHandler)
+	opts.SetPingTimeout(1 * time.Second)
+	opts.OnConnect = c.connectHandler
+	opts.OnConnectionLost = c.connectionLostHandler
+
+	// Create and start the client
+	c.client = mqtt.NewClient(opts)
+	if token := c.client.Connect(); token.Wait() && token.Error() != nil {
+		return fmt.Errorf("error connecting to MQTT broker: %v", token.Error())
+	}
+
+	// Subscribe to the configured topic
+	token := c.client.Subscribe(c.config.Topic, 0, nil)
+	token.Wait()
+	log.Printf("Subscribed to topic: %s\n", c.config.Topic)
+
+	return nil
+}
+
+// Disconnect cleanly disconnects from the MQTT broker
+func (c *Client) Disconnect() {
+	close(c.done)
+	token := c.client.Unsubscribe(c.config.Topic)
+	token.Wait()
+	c.client.Disconnect(250)
+}
+
+// Messages returns a channel of decoded messages
+// The consumer should read from this channel to receive decoded messages
+func (c *Client) Messages() <-chan DecodedMessage {
+	return c.decodedMessages
+}
+
+// These are intentionally left empty as we'll use the instance methods below
+
+// messageHandler processes incoming MQTT messages
+func (c *Client) messageHandler(client mqtt.Client, msg mqtt.Message) {
+	log.Printf("Received message from topic: %s\n", msg.Topic())
+
+	// Parse the topic structure
+	topicInfo, err := decoder.ParseTopic(msg.Topic())
+	if err != nil {
+		log.Printf("Error parsing topic: %v\n", err)
+		log.Printf("Raw topic: %s\n", msg.Topic())
+		log.Printf("Raw payload: %x\n", msg.Payload())
+		return
+	}
+
+	// For now, only handle "e" and "map" format messages
+	if topicInfo.Format == "e" || topicInfo.Format == "map" {
+		// Binary encoded protobuf message
+		decodedPacket := decoder.DecodeMessage(msg.Payload(), topicInfo)
+		
+		// Send the decoded message to the channel, but don't block if buffer is full
+		select {
+		case c.decodedMessages <- DecodedMessage{TopicInfo: topicInfo, DecodedPacket: decodedPacket}:
+			// Message sent successfully
+		case <-c.done:
+			// Client is shutting down
+			return
+		default:
+			// Channel buffer is full, log a warning and drop the message
+			log.Println("Warning: Message buffer full, dropping message")
+		}
+	} else if topicInfo.Format == "json" {
+		// TODO: Add support for JSON format messages in the future
+		log.Printf("Ignoring JSON format message from topic: %s\n", msg.Topic())
+	} else {
+		// Unsupported format, log and ignore
+		log.Printf("Unsupported format: %s from topic: %s\n", topicInfo.Format, msg.Topic())
+	}
+}
+
+// connectHandler is called when the client connects to the broker
+func (c *Client) connectHandler(client mqtt.Client) {
+	log.Println("Connected to MQTT Broker!")
+}
+
+// connectionLostHandler is called when the client loses connection
+func (c *Client) connectionLostHandler(client mqtt.Client, err error) {
+	log.Printf("Connection lost: %v\n", err)
+}
@@ -0,0 +1,69 @@
+package mqtt
+
+import (
+	"testing"
+	"time"
+)
+
+// TestClientConfig verifies that the client can be created with a config
+func TestClientConfig(t *testing.T) {
+	config := Config{
+		Broker:   "test.mosquitto.org",
+		Username: "test",
+		Password: "test",
+		ClientID: "test-client",
+		Topic:    "test/topic",
+	}
+
+	client := NewClient(config)
+	if client == nil {
+		t.Fatal("Expected client to be created, got nil")
+	}
+
+	if client.config.Broker != config.Broker {
+		t.Errorf("Expected broker to be %s, got %s", config.Broker, client.config.Broker)
+	}
+
+	if client.config.Topic != config.Topic {
+		t.Errorf("Expected topic to be %s, got %s", config.Topic, client.config.Topic)
+	}
+
+	// Check that channels are initialized
+	if client.decodedMessages == nil {
+		t.Error("Expected decodedMessages channel to be initialized")
+	}
+
+	if client.done == nil {
+		t.Error("Expected done channel to be initialized")
+	}
+
+	// Check buffer capacity
+	if cap(client.decodedMessages) != 100 {
+		t.Errorf("Expected decodedMessages buffer capacity to be 100, got %d", cap(client.decodedMessages))
+	}
+}
+
+// This test is a mock test and doesn't actually connect to MQTT
+// It just verifies that the messages channel works as expected
+func TestMessagesChannel(t *testing.T) {
+	client := NewClient(Config{})
+	ch := client.Messages()
+
+	// Verify we get the channel
+	if ch == nil {
+		t.Fatal("Expected messages channel, got nil")
+	}
+
+	// Test we can read from the channel
+	go func() {
+		msg := DecodedMessage{}
+		client.decodedMessages <- msg
+	}()
+
+	select {
+	case <-ch:
+		// Successfully received a message
+	case <-time.After(100 * time.Millisecond):
+		t.Fatal("Timed out waiting for message from channel")
+	}
+}