From 0784ccb95890a585dc0b8a91730f132fb0d50648 Mon Sep 17 00:00:00 2001 From: Daniel Pupius Date: Mon, 21 Apr 2025 22:34:49 -0700 Subject: [PATCH] Simplify packet construction --- .logista.yaml | 22 +++++++++---------- mqtt/broker.go | 23 ++++++++++---------- mqtt/broker_test.go | 52 ++++++++++++++++++--------------------------- mqtt/client.go | 10 ++++----- mqtt/client_test.go | 3 ++- mqtt/logger.go | 5 +++-- mqtt/packet.go | 14 +++++------- mqtt/stats.go | 3 ++- mqtt/subscriber.go | 19 +++++++++-------- 9 files changed, 71 insertions(+), 80 deletions(-) diff --git a/.logista.yaml b/.logista.yaml index efae45a..37b739c 100644 --- a/.logista.yaml +++ b/.logista.yaml @@ -12,21 +12,21 @@ format: | {{$timestamp }} {{ $level }} [{{ $logger }}] {{ $message }} {{if .packet -}} - {{- $packet := .packet -}} - {{"Channel:" | dim }} {{ $packet.ChannelID | color "green" }} - {{ "From:" | dim }} {{ $packet.From | mult 1 }} {{ "To:" | dim }} {{ $packet.To | mult 1 }} - {{ "Gateway:" | dim }} {{ $packet.GatewayID | color "yellow" }} - {{ "Priority:" | dim }} {{ $packet.Priority }} - {{ "PortNum:" | dim }} {{ $packet.PortNum }} + {{- $data := .packet.data -}} + {{"Channel:" | dim }} {{ $data.ChannelID | color "green" }} + {{ "From:" | dim }} {{ $data.From | mult 1 }} {{ "To:" | dim }} {{ $data.To | mult 1 }} + {{ "Gateway:" | dim }} {{ $data.GatewayID | color "yellow" }} + {{ "Priority:" | dim }} {{ $data.Priority }} + {{ "PortNum:" | dim }} {{ $data.PortNum }} - {{- if $packet.HopLimit }} - {{ "Hop:" | dim }} {{ $packet.HopStart }}/{{ $packet.HopLimit }} + {{- if $data.HopLimit }} + {{ "Hop:" | dim }} {{ $data.HopStart }}/{{ $data.HopLimit }} {{- end }} - {{- if $packet.DecodeError }} - {{ "Error:" | dim }} {{ $packet.DecodeError | color "red" }} + {{- if $data.DecodeError }} + {{ "Error:" | dim }} {{ $data.DecodeError | color "red" }} {{- end }} {{ "Payload:" | dim }} - {{ $packet.Payload | table }} + {{ $data.Payload | table }} {{end -}} diff --git a/mqtt/broker.go b/mqtt/broker.go index 0e2ae1b..390e99b 100644 --- a/mqtt/broker.go +++ b/mqtt/broker.go @@ -4,12 +4,13 @@ import ( "sync" "github.com/dpup/prefab/logging" + meshtreampb "meshstream/generated/meshstream" ) // Broker distributes messages from a source channel to multiple subscriber channels type Broker struct { - sourceChan <-chan *Packet // Source of packets (e.g., from MQTT client) - subscribers map[chan *Packet]struct{} // Active subscribers + sourceChan <-chan *meshtreampb.Packet // Source of packets (e.g., from MQTT client) + subscribers map[chan *meshtreampb.Packet]struct{} // Active subscribers subscriberMutex sync.RWMutex // Lock for modifying the subscribers map done chan struct{} // Signal to stop the dispatch loop wg sync.WaitGroup // Wait group to ensure clean shutdown @@ -17,10 +18,10 @@ type Broker struct { } // NewBroker creates a new broker that distributes messages from sourceChannel to subscribers -func NewBroker(sourceChannel <-chan *Packet, logger logging.Logger) *Broker { +func NewBroker(sourceChannel <-chan *meshtreampb.Packet, logger logging.Logger) *Broker { broker := &Broker{ sourceChan: sourceChannel, - subscribers: make(map[chan *Packet]struct{}), + subscribers: make(map[chan *meshtreampb.Packet]struct{}), done: make(chan struct{}), logger: logger.Named("mqtt.broker"), } @@ -34,9 +35,9 @@ func NewBroker(sourceChannel <-chan *Packet, logger logging.Logger) *Broker { // Subscribe creates and returns a new subscriber channel // The bufferSize parameter controls how many messages can be buffered in the channel -func (b *Broker) Subscribe(bufferSize int) <-chan *Packet { +func (b *Broker) Subscribe(bufferSize int) <-chan *meshtreampb.Packet { // Create a new channel for this subscriber - subscriberChan := make(chan *Packet, bufferSize) + subscriberChan := make(chan *meshtreampb.Packet, bufferSize) // Register the new subscriber b.subscriberMutex.Lock() @@ -48,7 +49,7 @@ func (b *Broker) Subscribe(bufferSize int) <-chan *Packet { } // Unsubscribe removes a subscriber and closes its channel -func (b *Broker) Unsubscribe(ch <-chan *Packet) { +func (b *Broker) Unsubscribe(ch <-chan *meshtreampb.Packet) { b.subscriberMutex.Lock() defer b.subscriberMutex.Unlock() @@ -80,7 +81,7 @@ func (b *Broker) Close() { for ch := range b.subscribers { close(ch) } - b.subscribers = make(map[chan *Packet]struct{}) + b.subscribers = make(map[chan *meshtreampb.Packet]struct{}) } // dispatchLoop continuously reads from the source channel and distributes to subscribers @@ -108,10 +109,10 @@ func (b *Broker) dispatchLoop() { } // broadcast sends a packet to all active subscribers without blocking -func (b *Broker) broadcast(packet *Packet) { +func (b *Broker) broadcast(packet *meshtreampb.Packet) { // Take a read lock to get a snapshot of the subscribers b.subscriberMutex.RLock() - subscribers := make([]chan *Packet, 0, len(b.subscribers)) + subscribers := make([]chan *meshtreampb.Packet, 0, len(b.subscribers)) for ch := range b.subscribers { subscribers = append(subscribers, ch) } @@ -120,7 +121,7 @@ func (b *Broker) broadcast(packet *Packet) { // Distribute to all subscribers for _, ch := range subscribers { // Use a goroutine and recover to ensure sending to a closed channel doesn't panic - go func(ch chan *Packet) { + go func(ch chan *meshtreampb.Packet) { defer func() { if r := recover(); r != nil { // This can happen if the channel was closed after we took a snapshot diff --git a/mqtt/broker_test.go b/mqtt/broker_test.go index 76d6df0..43ec655 100644 --- a/mqtt/broker_test.go +++ b/mqtt/broker_test.go @@ -13,7 +13,7 @@ import ( // TestBrokerSubscribeUnsubscribe tests the basic subscribe and unsubscribe functionality func TestBrokerSubscribeUnsubscribe(t *testing.T) { // Create a test source channel - sourceChan := make(chan *Packet, 10) + sourceChan := make(chan *meshtreampb.Packet, 10) // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") @@ -37,11 +37,9 @@ func TestBrokerSubscribeUnsubscribe(t *testing.T) { // and exact packet matching may not work reliably // First packet with ID 1 - packet1 := &Packet{ - Packet: &meshtreampb.Packet{ - Data: &meshtreampb.Data{Id: 1}, - Info: &meshtreampb.TopicInfo{}, - }, + packet1 := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: 1}, + Info: &meshtreampb.TopicInfo{}, } // Send the packet @@ -79,11 +77,9 @@ func TestBrokerSubscribeUnsubscribe(t *testing.T) { } // Second packet with ID 2 - packet2 := &Packet{ - Packet: &meshtreampb.Packet{ - Data: &meshtreampb.Data{Id: 2}, - Info: &meshtreampb.TopicInfo{}, - }, + packet2 := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: 2}, + Info: &meshtreampb.TopicInfo{}, } // Send the second packet @@ -103,7 +99,7 @@ func TestBrokerSubscribeUnsubscribe(t *testing.T) { // TestBrokerMultipleSubscribers tests broadcasting to many subscribers func TestBrokerMultipleSubscribers(t *testing.T) { // Create a test source channel - sourceChan := make(chan *Packet, 10) + sourceChan := make(chan *meshtreampb.Packet, 10) // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") @@ -112,17 +108,15 @@ func TestBrokerMultipleSubscribers(t *testing.T) { // Create multiple subscribers const numSubscribers = 10 - subscribers := make([]<-chan *Packet, numSubscribers) + subscribers := make([]<-chan *meshtreampb.Packet, numSubscribers) for i := 0; i < numSubscribers; i++ { subscribers[i] = broker.Subscribe(5) } // Send a test packet with ID 42 - testPacket := &Packet{ - Packet: &meshtreampb.Packet{ - Data: &meshtreampb.Data{Id: 42}, - Info: &meshtreampb.TopicInfo{}, - }, + testPacket := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: 42}, + Info: &meshtreampb.TopicInfo{}, } sourceChan <- testPacket @@ -131,7 +125,7 @@ func TestBrokerMultipleSubscribers(t *testing.T) { wg.Add(numSubscribers) for i, subscriber := range subscribers { - go func(idx int, ch <-chan *Packet) { + go func(idx int, ch <-chan *meshtreampb.Packet) { defer wg.Done() select { case received := <-ch: @@ -151,7 +145,7 @@ func TestBrokerMultipleSubscribers(t *testing.T) { // TestBrokerSlowSubscriber tests that a slow subscriber doesn't block others func TestBrokerSlowSubscriber(t *testing.T) { // Create a test source channel - sourceChan := make(chan *Packet, 10) + sourceChan := make(chan *meshtreampb.Packet, 10) // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") @@ -174,17 +168,13 @@ func TestBrokerSlowSubscriber(t *testing.T) { } // Send two packets quickly to fill the slow subscriber's buffer - testPacket1 := &Packet{ - Packet: &meshtreampb.Packet{ - Data: &meshtreampb.Data{Id: 101}, - Info: &meshtreampb.TopicInfo{}, - }, + testPacket1 := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: 101}, + Info: &meshtreampb.TopicInfo{}, } - testPacket2 := &Packet{ - Packet: &meshtreampb.Packet{ - Data: &meshtreampb.Data{Id: 102}, - Info: &meshtreampb.TopicInfo{}, - }, + testPacket2 := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: 102}, + Info: &meshtreampb.TopicInfo{}, } sourceChan <- testPacket1 @@ -227,7 +217,7 @@ func TestBrokerSlowSubscriber(t *testing.T) { // TestBrokerCloseWithSubscribers tests closing the broker with active subscribers func TestBrokerCloseWithSubscribers(t *testing.T) { // Create a test source channel - sourceChan := make(chan *Packet, 10) + sourceChan := make(chan *meshtreampb.Packet, 10) // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") diff --git a/mqtt/client.go b/mqtt/client.go index 9e4de5e..e7dcc5a 100644 --- a/mqtt/client.go +++ b/mqtt/client.go @@ -8,6 +8,7 @@ import ( mqtt "github.com/eclipse/paho.mqtt.golang" "meshstream/decoder" + meshtreampb "meshstream/generated/meshstream" ) // Config holds configuration for the MQTT client @@ -23,7 +24,7 @@ type Config struct { type Client struct { config Config client mqtt.Client - decodedMessages chan *Packet + decodedMessages chan *meshtreampb.Packet done chan struct{} logger logging.Logger } @@ -32,7 +33,7 @@ type Client struct { func NewClient(config Config, logger logging.Logger) *Client { return &Client{ config: config, - decodedMessages: make(chan *Packet, 100), + decodedMessages: make(chan *meshtreampb.Packet, 100), done: make(chan struct{}), logger: logger.Named("mqtt.client"), } @@ -40,7 +41,6 @@ func NewClient(config Config, logger logging.Logger) *Client { // 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) @@ -75,7 +75,7 @@ func (c *Client) Disconnect() { // Messages returns a channel of decoded messages // The consumer should read from this channel to receive decoded messages -func (c *Client) Messages() <-chan *Packet { +func (c *Client) Messages() <-chan *meshtreampb.Packet { return c.decodedMessages } @@ -133,4 +133,4 @@ func (c *Client) connectHandler(client mqtt.Client) { // connectionLostHandler is called when the client loses connection func (c *Client) connectionLostHandler(client mqtt.Client, err error) { c.logger.Errorw("Connection lost", "error", err) -} \ No newline at end of file +} diff --git a/mqtt/client_test.go b/mqtt/client_test.go index 62d8f47..309f98f 100644 --- a/mqtt/client_test.go +++ b/mqtt/client_test.go @@ -5,6 +5,7 @@ import ( "time" "github.com/dpup/prefab/logging" + meshtreampb "meshstream/generated/meshstream" ) // TestClientConfig verifies that the client can be created with a config @@ -60,7 +61,7 @@ func TestMessagesChannel(t *testing.T) { // Test we can read from the channel go func() { - msg := &Packet{} + msg := &meshtreampb.Packet{} client.decodedMessages <- msg }() diff --git a/mqtt/logger.go b/mqtt/logger.go index 3cf8816..4ebb6b6 100644 --- a/mqtt/logger.go +++ b/mqtt/logger.go @@ -7,6 +7,7 @@ import ( "github.com/dpup/prefab/logging" pb "meshstream/generated/meshtastic" + meshtreampb "meshstream/generated/meshstream" ) // MessageLogger logs messages using the provided logger @@ -41,7 +42,7 @@ func NewMessageLogger(broker *Broker, briefMode bool, logger logging.Logger) (*M } // getBriefSummary returns a brief summary of the packet -func (ml *MessageLogger) getBriefSummary(packet *Packet) string { +func (ml *MessageLogger) getBriefSummary(packet *meshtreampb.Packet) string { var summary string if packet.Data.DecodeError != "" { @@ -95,7 +96,7 @@ func (ml *MessageLogger) getBriefSummary(packet *Packet) string { } // logMessage logs a message using the structured logger -func (ml *MessageLogger) logMessage(packet *Packet) { +func (ml *MessageLogger) logMessage(packet *meshtreampb.Packet) { // Get a brief summary for structured logging briefSummary := ml.getBriefSummary(packet) diff --git a/mqtt/packet.go b/mqtt/packet.go index bfd58e4..14e6c0b 100644 --- a/mqtt/packet.go +++ b/mqtt/packet.go @@ -4,14 +4,10 @@ import ( meshtreampb "meshstream/generated/meshstream" ) -// Type alias for proto packet. -type Packet meshtreampb.Packet - -// NewPacket creates a Packet from a data packet and topic info -func NewPacket(data *meshtreampb.Data, topicInfo *meshtreampb.TopicInfo) *Packet { - p := Packet(meshtreampb.Packet{ +// NewPacket creates a new Packet from a data packet and topic info +func NewPacket(data *meshtreampb.Data, topicInfo *meshtreampb.TopicInfo) *meshtreampb.Packet { + return &meshtreampb.Packet{ Data: data, Info: topicInfo, - }) - return &p -} + } +} \ No newline at end of file diff --git a/mqtt/stats.go b/mqtt/stats.go index 88e49ac..9c4b247 100644 --- a/mqtt/stats.go +++ b/mqtt/stats.go @@ -8,6 +8,7 @@ import ( "github.com/dpup/prefab/logging" pb "meshstream/generated/meshtastic" + meshtreampb "meshstream/generated/meshstream" ) // MessageStats tracks statistics about received messages @@ -64,7 +65,7 @@ func (s *MessageStats) runTicker() { } // recordMessage records a message in the statistics -func (s *MessageStats) recordMessage(packet *Packet) { +func (s *MessageStats) recordMessage(packet *meshtreampb.Packet) { s.Lock() defer s.Unlock() diff --git a/mqtt/subscriber.go b/mqtt/subscriber.go index b9fd5c4..924ff8b 100644 --- a/mqtt/subscriber.go +++ b/mqtt/subscriber.go @@ -4,27 +4,28 @@ import ( "sync" "github.com/dpup/prefab/logging" + meshtreampb "meshstream/generated/meshstream" ) // SubscriberConfig holds configuration for creating a subscriber type SubscriberConfig struct { - Name string // Descriptive name for the subscriber - Broker *Broker // The broker to subscribe to - BufferSize int // Channel buffer size - Processor func(*Packet) // Function to process each packet - StartHook func() // Optional hook called when starting - CloseHook func() // Optional hook called when closing - Logger logging.Logger // Logger instance to use + Name string // Descriptive name for the subscriber + Broker *Broker // The broker to subscribe to + BufferSize int // Channel buffer size + Processor func(*meshtreampb.Packet) // Function to process each packet + StartHook func() // Optional hook called when starting + CloseHook func() // Optional hook called when closing + Logger logging.Logger // Logger instance to use } // BaseSubscriber implements common subscriber functionality type BaseSubscriber struct { broker *Broker - channel <-chan *Packet + channel <-chan *meshtreampb.Packet done chan struct{} wg sync.WaitGroup name string - processor func(*Packet) + processor func(*meshtreampb.Packet) startHook func() closeHook func() BufferSize int