From 879877fa5313222af4cad1dac182aa836f8d3668 Mon Sep 17 00:00:00 2001 From: Daniel Pupius Date: Sun, 20 Apr 2025 18:11:15 -0700 Subject: [PATCH] Refactor MQTT handling into separate package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- main.go | 124 ++++++++++++++-------------------------- mqtt/client.go | 134 ++++++++++++++++++++++++++++++++++++++++++++ mqtt/client_test.go | 69 +++++++++++++++++++++++ 3 files changed, 244 insertions(+), 83 deletions(-) create mode 100644 mqtt/client.go create mode 100644 mqtt/client_test.go diff --git a/main.go b/main.go index db668aa..be47493 100644 --- a/main.go +++ b/main.go @@ -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 + } + } +} \ No newline at end of file diff --git a/mqtt/client.go b/mqtt/client.go new file mode 100644 index 0000000..2f207cb --- /dev/null +++ b/mqtt/client.go @@ -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) +} \ No newline at end of file diff --git a/mqtt/client_test.go b/mqtt/client_test.go new file mode 100644 index 0000000..1204ef6 --- /dev/null +++ b/mqtt/client_test.go @@ -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") + } +} \ No newline at end of file