diff --git a/main.go b/main.go index 3acbb63..16cb2c1 100644 --- a/main.go +++ b/main.go @@ -58,7 +58,8 @@ func main() { messagesChan := mqttClient.Messages() // Create a message broker to distribute messages to multiple consumers - broker := mqtt.NewBroker(messagesChan, logger) + // Cache the last 50 packets for new subscribers + broker := mqtt.NewBroker(messagesChan, 50, logger) // Create a stats tracker that subscribes to the broker // with statistics printed every 30 seconds diff --git a/mqtt/broker.go b/mqtt/broker.go index 390e99b..6cb3af2 100644 --- a/mqtt/broker.go +++ b/mqtt/broker.go @@ -7,23 +7,90 @@ import ( meshtreampb "meshstream/generated/meshstream" ) +// CircularBuffer implements a fixed-size circular buffer for caching packets +type CircularBuffer struct { + buffer []*meshtreampb.Packet // Fixed size buffer to store packets + size int // Size of the buffer + next int // Index where the next packet will be stored + count int // Number of packets currently in the buffer + mutex sync.RWMutex // Lock for thread-safe access +} + +// NewCircularBuffer creates a new circular buffer with the given size +func NewCircularBuffer(size int) *CircularBuffer { + return &CircularBuffer{ + buffer: make([]*meshtreampb.Packet, size), + size: size, + next: 0, + count: 0, + } +} + +// Add adds a packet to the circular buffer +func (cb *CircularBuffer) Add(packet *meshtreampb.Packet) { + cb.mutex.Lock() + defer cb.mutex.Unlock() + + cb.buffer[cb.next] = packet + cb.next = (cb.next + 1) % cb.size + + // Update count if we haven't filled the buffer yet + if cb.count < cb.size { + cb.count++ + } +} + +// GetAll returns all packets in the buffer in chronological order +func (cb *CircularBuffer) GetAll() []*meshtreampb.Packet { + cb.mutex.RLock() + defer cb.mutex.RUnlock() + + if cb.count == 0 { + return []*meshtreampb.Packet{} + } + + result := make([]*meshtreampb.Packet, cb.count) + + // If buffer isn't full yet, just copy from start to next + if cb.count < cb.size { + copy(result, cb.buffer[:cb.count]) + return result + } + + // If buffer is full, we need to handle the wrap-around + // First copy from next to end + firstPartLength := cb.size - cb.next + if firstPartLength > 0 { + copy(result, cb.buffer[cb.next:]) + } + + // Then copy from start to next + if cb.next > 0 { + copy(result[firstPartLength:], cb.buffer[:cb.next]) + } + + return result +} + // Broker distributes messages from a source channel to multiple subscriber channels type Broker struct { 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 - logger logging.Logger // Logger for broker operations + 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 + logger logging.Logger // Logger for broker operations + cache *CircularBuffer // Circular buffer for caching packets } // NewBroker creates a new broker that distributes messages from sourceChannel to subscribers -func NewBroker(sourceChannel <-chan *meshtreampb.Packet, logger logging.Logger) *Broker { +func NewBroker(sourceChannel <-chan *meshtreampb.Packet, cacheSize int, logger logging.Logger) *Broker { broker := &Broker{ sourceChan: sourceChannel, subscribers: make(map[chan *meshtreampb.Packet]struct{}), done: make(chan struct{}), logger: logger.Named("mqtt.broker"), + cache: NewCircularBuffer(cacheSize), } // Start the dispatch loop @@ -44,6 +111,23 @@ func (b *Broker) Subscribe(bufferSize int) <-chan *meshtreampb.Packet { b.subscribers[subscriberChan] = struct{}{} b.subscriberMutex.Unlock() + // Send cached packets to the new subscriber + cachedPackets := b.cache.GetAll() + if len(cachedPackets) > 0 { + go func() { + for _, packet := range cachedPackets { + select { + case subscriberChan <- packet: + // Successfully sent packet + default: + // Buffer is full, log warning and stop sending cached packets + b.logger.Warn("New subscriber buffer full, stopping cache replay") + return + } + } + }() + } + // Return the channel return subscriberChan } @@ -102,6 +186,9 @@ func (b *Broker) dispatchLoop() { return } + // Add packet to the cache + b.cache.Add(packet) + // Distribute the packet to all subscribers b.broadcast(packet) } @@ -139,4 +226,4 @@ func (b *Broker) broadcast(packet *meshtreampb.Packet) { } }(ch) } -} +} \ No newline at end of file diff --git a/mqtt/broker_test.go b/mqtt/broker_test.go index 43ec655..be4811d 100644 --- a/mqtt/broker_test.go +++ b/mqtt/broker_test.go @@ -10,6 +10,64 @@ import ( meshtreampb "meshstream/generated/meshstream" ) +// TestCircularBuffer tests the circular buffer implementation +func TestCircularBuffer(t *testing.T) { + // Create a circular buffer with size 3 + buffer := NewCircularBuffer(3) + + // Test empty buffer returns empty slice + packets := buffer.GetAll() + if len(packets) != 0 { + t.Errorf("Expected empty buffer to return empty slice, got %d items", len(packets)) + } + + // Add 3 packets and verify count + for i := 1; i <= 3; i++ { + packet := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: uint32(i)}, + Info: &meshtreampb.TopicInfo{}, + } + buffer.Add(packet) + } + + // Check that buffer has 3 packets + packets = buffer.GetAll() + if len(packets) != 3 { + t.Errorf("Expected buffer to have 3 packets, got %d", len(packets)) + } + + // Verify packets are in order + for i, packet := range packets { + expected := uint32(i + 1) + if packet.Data.Id != expected { + t.Errorf("Expected packet %d to have ID %d, got %d", i, expected, packet.Data.Id) + } + } + + // Add 2 more packets to test wrap-around + for i := 4; i <= 5; i++ { + packet := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: uint32(i)}, + Info: &meshtreampb.TopicInfo{}, + } + buffer.Add(packet) + } + + // Check that buffer still has 3 packets (maxed out) + packets = buffer.GetAll() + if len(packets) != 3 { + t.Errorf("Expected buffer to have 3 packets after overflow, got %d", len(packets)) + } + + // Verify packets are the latest ones in order (3, 4, 5) + for i, packet := range packets { + expected := uint32(i + 3) + if packet.Data.Id != expected { + t.Errorf("Expected packet %d to have ID %d, got %d", i, expected, packet.Data.Id) + } + } +} + // TestBrokerSubscribeUnsubscribe tests the basic subscribe and unsubscribe functionality func TestBrokerSubscribeUnsubscribe(t *testing.T) { // Create a test source channel @@ -17,7 +75,7 @@ func TestBrokerSubscribeUnsubscribe(t *testing.T) { // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") - broker := NewBroker(sourceChan, testLogger) + broker := NewBroker(sourceChan, 5, testLogger) defer broker.Close() // Subscribe to the broker @@ -103,7 +161,7 @@ func TestBrokerMultipleSubscribers(t *testing.T) { // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") - broker := NewBroker(sourceChan, testLogger) + broker := NewBroker(sourceChan, 5, testLogger) defer broker.Close() // Create multiple subscribers @@ -149,7 +207,7 @@ func TestBrokerSlowSubscriber(t *testing.T) { // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") - broker := NewBroker(sourceChan, testLogger) + broker := NewBroker(sourceChan, 5, testLogger) defer broker.Close() // Create a slow subscriber with buffer size 1 @@ -221,7 +279,7 @@ func TestBrokerCloseWithSubscribers(t *testing.T) { // Create a broker with the source channel testLogger := logging.NewDevLogger().Named("test") - broker := NewBroker(sourceChan, testLogger) + broker := NewBroker(sourceChan, 5, testLogger) // Subscribe to the broker subscriber := broker.Subscribe(5) @@ -248,4 +306,181 @@ func TestBrokerCloseWithSubscribers(t *testing.T) { case <-time.After(100 * time.Millisecond): t.Error("Subscriber channel should be closed but isn't") } +} + +// TestBrokerPacketCaching tests that the broker caches packets +func TestBrokerPacketCaching(t *testing.T) { + // Create a test source channel + sourceChan := make(chan *meshtreampb.Packet, 10) + + // Create a broker with a small cache size + testLogger := logging.NewDevLogger().Named("test") + broker := NewBroker(sourceChan, 3, testLogger) + defer broker.Close() + + // Send three packets + for i := 1; i <= 3; i++ { + packet := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: uint32(i)}, + Info: &meshtreampb.TopicInfo{}, + } + sourceChan <- packet + + // Give the broker time to process the packet + time.Sleep(10 * time.Millisecond) + } + + // Create a subscriber after the packets were sent + subscriber := broker.Subscribe(5) + + // The subscriber should receive all three cached packets + receivedIds := make([]uint32, 0, 3) + + // We need to receive 3 packets + for i := 0; i < 3; i++ { + select { + case received := <-subscriber: + receivedIds = append(receivedIds, received.Data.Id) + case <-time.After(100 * time.Millisecond): + t.Errorf("Subscriber didn't receive cached packet %d within timeout", i+1) + } + } + + // Check that we received all packets in the correct order + if len(receivedIds) != 3 { + t.Errorf("Expected to receive 3 packets, got %d", len(receivedIds)) + } else { + for i, id := range receivedIds { + if id != uint32(i+1) { + t.Errorf("Expected packet %d to have ID %d, got %d", i, i+1, id) + } + } + } +} + +// TestBrokerCacheOverflow tests that the broker correctly handles cache overflow +func TestBrokerCacheOverflow(t *testing.T) { + // Create a test source channel + sourceChan := make(chan *meshtreampb.Packet, 10) + + // Create a broker with a small cache size + testLogger := logging.NewDevLogger().Named("test") + cacheSize := 3 + broker := NewBroker(sourceChan, cacheSize, testLogger) + defer broker.Close() + + // Send 5 packets (exceeding the cache size) + for i := 1; i <= 5; i++ { + packet := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: uint32(i)}, + Info: &meshtreampb.TopicInfo{}, + } + sourceChan <- packet + + // Give the broker time to process the packet + time.Sleep(10 * time.Millisecond) + } + + // Create a subscriber after the packets were sent + subscriber := broker.Subscribe(5) + + // The subscriber should receive the last 3 packets (3, 4, 5) + receivedIds := make([]uint32, 0, cacheSize) + + // We expect to receive exactly cacheSize packets + for i := 0; i < cacheSize; i++ { + select { + case received := <-subscriber: + receivedIds = append(receivedIds, received.Data.Id) + case <-time.After(100 * time.Millisecond): + t.Errorf("Subscriber didn't receive cached packet %d within timeout", i+1) + } + } + + // Verify no more packets are coming + select { + case received := <-subscriber: + t.Errorf("Received unexpected packet with ID %d", received.Data.Id) + case <-time.After(50 * time.Millisecond): + // This is expected, no more packets should be received + } + + // Check that we received only the last 3 packets in the correct order + expectedIds := []uint32{3, 4, 5} + if len(receivedIds) != len(expectedIds) { + t.Errorf("Expected to receive %d packets, got %d", len(expectedIds), len(receivedIds)) + } else { + for i, id := range receivedIds { + if id != expectedIds[i] { + t.Errorf("Expected packet %d to have ID %d, got %d", i, expectedIds[i], id) + } + } + } +} + +// TestSubscriberBufferFull tests the behavior when a subscriber's buffer is full +func TestSubscriberBufferFull(t *testing.T) { + // Create a test source channel + sourceChan := make(chan *meshtreampb.Packet, 10) + + // Create a broker with a cache size of 5 + testLogger := logging.NewDevLogger().Named("test") + broker := NewBroker(sourceChan, 5, testLogger) + defer broker.Close() + + // Prefill the cache with 5 packets + for i := 1; i <= 5; i++ { + packet := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: uint32(i)}, + Info: &meshtreampb.TopicInfo{}, + } + sourceChan <- packet + time.Sleep(10 * time.Millisecond) + } + + // Create a subscriber with a very small buffer size (1) + smallSubscriber := broker.Subscribe(1) + + // The small subscriber should receive at least one cached packet + // The others will be dropped because the buffer is full + select { + case received := <-smallSubscriber: + if received.Data.Id != 1 { + t.Errorf("Expected subscriber to receive packet with ID 1, got %d", received.Data.Id) + } + case <-time.After(100 * time.Millisecond): + t.Error("Subscriber didn't receive any cached packet within timeout") + } + + // Check that no more packets are immediately available + // This is a bit tricky to test since we can't guarantee how many + // packets were dropped due to the full buffer, but we can check + // that the channel isn't immediately ready with more packets + select { + case received := <-smallSubscriber: + // If we get here, it should be a later packet from the cache + if received.Data.Id < 1 { + t.Errorf("Received unexpected packet with ID %d", received.Data.Id) + } + case <-time.After(50 * time.Millisecond): + // This is also acceptable - it means all attempts to send more cached + // packets found the buffer full and gave up + } + + // Send a new packet now that the subscriber is connected + newPacket := &meshtreampb.Packet{ + Data: &meshtreampb.Data{Id: 6}, + Info: &meshtreampb.TopicInfo{}, + } + sourceChan <- newPacket + + // The subscriber should receive this packet if they read the first one + select { + case received := <-smallSubscriber: + if received.Data.Id != 6 { + t.Errorf("Expected subscriber to receive packet with ID 6, got %d", received.Data.Id) + } + case <-time.After(100 * time.Millisecond): + t.Error("Subscriber didn't receive new packet within timeout") + } } \ No newline at end of file diff --git a/server/server.go b/server/server.go index e12f130..4e440c0 100644 --- a/server/server.go +++ b/server/server.go @@ -133,8 +133,8 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) { return } - // Subscribe to the broker with a buffer size of 10 - packetChan := s.config.Broker.Subscribe(10) + // Subscribe to the broker with a buffer size of 100 + packetChan := s.config.Broker.Subscribe(100) // Signal when the client disconnects notify := ctx.Done() diff --git a/web/src/components/Nav.tsx b/web/src/components/Nav.tsx index da9d183..c514244 100644 --- a/web/src/components/Nav.tsx +++ b/web/src/components/Nav.tsx @@ -28,7 +28,7 @@ interface NavProps { // Navigation items data const navigationItems: NavItem[] = [ { - to: "/", + to: "/home", label: "Dashboard", icon: LayoutDashboard, exact: true, @@ -77,9 +77,11 @@ export const Nav: React.FC = ({ connectionStatus }) => { "text-neutral-400 hover:text-neutral-200 font-thin", }} activeProps={{ - exact: item.exact, className: "text-neutral-200 font-normal", }} + activeOptions={{ + exact: item.exact, + }} > {item.label} diff --git a/web/src/components/PacketList.tsx b/web/src/components/PacketList.tsx index 4ff77a9..e9a7bde 100644 --- a/web/src/components/PacketList.tsx +++ b/web/src/components/PacketList.tsx @@ -103,13 +103,9 @@ export const PacketList: React.FC = () => { }; return ( -
+
-
- {packets.length} packets received - {packets.length > 0 && <>, since {getEarliestTime()}} -
{/* Show buffered count when paused */} {streamPaused && bufferedPackets.length > 0 && ( @@ -134,6 +130,10 @@ export const PacketList: React.FC = () => { Clear
+
+ {packets.length} packets received + {packets.length > 0 && <>, since {getEarliestTime()}} +
@@ -175,11 +175,17 @@ export const PacketList: React.FC = () => {
{/* Empty state when no packets are visible but stream is paused */} {packets.length === 0 && streamPaused && bufferedPackets.length > 0 && ( -
- Stream is paused with {bufferedPackets.length} buffered messages. +
+
+ + + Stream is paused with {bufferedPackets.length} buffered + messages. + +
@@ -188,41 +194,46 @@ export const PacketList: React.FC = () => { {/* Pagination */} {totalPages > 1 && ( -
- + <> + +
+
+ Page {currentPage} of {totalPages} +
-
- Page {currentPage} of {totalPages} +
+ + + +
- - -
+ )}
diff --git a/web/src/components/dashboard/NodeDetail.tsx b/web/src/components/dashboard/NodeDetail.tsx index caa3060..583b1d3 100644 --- a/web/src/components/dashboard/NodeDetail.tsx +++ b/web/src/components/dashboard/NodeDetail.tsx @@ -1,10 +1,11 @@ -import React, { useEffect, useRef } from "react"; +import React, { useEffect, useRef, useCallback } from "react"; import { useNavigate, Link } from "@tanstack/react-router"; import { useAppSelector, useAppDispatch } from "../../hooks"; import { selectNode } from "../../store/slices/aggregatorSlice"; -import { RegionCode, ModemPreset } from "../../lib/types"; +import { RegionCode, ModemPreset, Packet } from "../../lib/types"; import { KeyValuePair } from "../ui/KeyValuePair"; import { Separator } from "../Separator"; +import { PacketRenderer } from "../packets/PacketRenderer"; import { ArrowLeft, Radio, @@ -29,6 +30,7 @@ import { Earth, TableConfig, Save, + MessageSquare, } from "lucide-react"; interface NodeDetailProps { @@ -357,6 +359,63 @@ const getModemPresetName = ( return presetNames[preset] || `Unknown (${preset})`; }; +// Component to render packets associated with a specific node +const NodePacketList: React.FC<{ nodeId: number }> = ({ nodeId }) => { + const { packets } = useAppSelector((state) => state.packets); + // Fixed number of packets to display + const MAX_PACKETS = 20; + + // Get packets from this node (sent or received) + const nodePackets = packets + .filter( + (packet) => packet.data.from === nodeId || packet.data.to === nodeId + ) + .slice(0, MAX_PACKETS); // Show fixed number of packets + + // Generate a reproducible packet key + const getPacketKey = useCallback((packet: Packet, index: number): string => { + if (packet.data.id !== undefined && packet.data.from !== undefined) { + const fromId = `!${packet.data.from.toString(16).toLowerCase()}`; + return `${fromId}_${packet.data.id}`; + } + return `fallback_${index}`; + }, []); + + if (nodePackets.length === 0) { + return ( +
+ No packets found for this node +
+ ); + } + + return ( +
+
+

+ Showing {nodePackets.length} of{" "} + { + packets.filter( + (p) => p.data.from === nodeId || p.data.to === nodeId + ).length + }{" "} + recent packets +

+
+ + + +
    + {nodePackets.map((packet, index) => ( +
  • + +
    • + ))} +
+
+ ); +}; + export const NodeDetail: React.FC = ({ nodeId }) => { const dispatch = useAppDispatch(); const navigate = useNavigate(); @@ -481,7 +540,7 @@ export const NodeDetail: React.FC = ({ nodeId }) => { const positionAccuracy = calculateAccuracyFromPrecisionBits(precisionBits); return ( -
+
{/* Header with back button and basic node info */}