adam/meshstream
1
0
Fork 0
mirror of https://github.com/dpup/meshstream.git synced 2026-07-31 22:12:40 +02:00
Code Issues Packages Projects Releases Wiki Activity

Simplify packet construction

Browse Source
This commit is contained in:
Daniel Pupius
2025-04-21 22:34:49 -07:00
parent f580249162
commit 0784ccb958
9 changed files with 71 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.logista.yaml
+11 -11
View File
@@ -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 -}}
mqtt/broker.go
+12 -11
View File
@@ -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
mqtt/broker_test.go
+21 -31
View File
@@ -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")
mqtt/client.go
+5 -5
View File
@@ -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)
}
}
mqtt/client_test.go
+2 -1
View File
@@ -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
}()
mqtt/logger.go
+3 -2
View File
@@ -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)
mqtt/packet.go
+5 -9
View File
@@ -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
}
}
}
mqtt/stats.go
+2 -1
View File
@@ -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()
mqtt/subscriber.go
+10 -9
View File
@@ -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