From 94c0f1a48c1b7b2b0710a5c3e642cd7616a6f8d1 Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 28 Jul 2025 13:12:06 -0500 Subject: [PATCH] Fix real-time packet updates not showing on map MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The map wasn't receiving real-time packet updates because: 1. LiveView wasn't subscribed to StreamingPacketsPubSub 2. LiveView wasn't handling {:streaming_packet, packet} messages 3. Packets weren't being broadcast to both PubSub systems Changes: - Added StreamingPacketsPubSub subscription with viewport bounds filtering - Added handler for {:streaming_packet, packet} messages - Update subscription bounds when map viewport changes - Properly unsubscribe on LiveView termination - Also broadcast packets to SpatialPubSub (in addition to StreamingPacketsPubSub) Now the map only receives packets within its current viewport bounds, reducing unnecessary data transfer and improving performance. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- lib/aprsme/packet_consumer.ex | 2 ++ lib/aprsme_web/live/map_live/index.ex | 11 +++++++++++ 2 files changed, 13 insertions(+) diff --git a/lib/aprsme/packet_consumer.ex b/lib/aprsme/packet_consumer.ex index 7380074..575619e 100644 --- a/lib/aprsme/packet_consumer.ex +++ b/lib/aprsme/packet_consumer.ex @@ -272,6 +272,8 @@ defmodule Aprsme.PacketConsumer do else # Normal single-node broadcasting Aprsme.StreamingPacketsPubSub.broadcast_packet(packet) + # Also broadcast to SpatialPubSub for viewport-based filtering + Aprsme.SpatialPubSub.broadcast_packet(packet) end end) rescue diff --git a/lib/aprsme_web/live/map_live/index.ex b/lib/aprsme_web/live/map_live/index.ex index 018b473..68da3c1 100644 --- a/lib/aprsme_web/live/map_live/index.ex +++ b/lib/aprsme_web/live/map_live/index.ex @@ -107,6 +107,9 @@ defmodule AprsmeWeb.MapLive.Index do # Still subscribe to bad packets (they don't have location) Phoenix.PubSub.subscribe(Aprsme.PubSub, "bad_packets") + # Subscribe to StreamingPacketsPubSub with initial bounds + Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), default_bounds) + Process.send_after(self(), :cleanup_old_packets, 60_000) # Schedule UI update timer to refresh "time ago" display every 30 seconds Process.send_after(self(), :update_time_display, 30_000) @@ -666,6 +669,8 @@ defmodule AprsmeWeb.MapLive.Index do def handle_info({:spatial_packet, packet}, socket), do: handle_info_postgres_packet(packet, socket) + def handle_info({:streaming_packet, packet}, socket), do: handle_info_postgres_packet(packet, socket) + def handle_info({:load_rf_path_station_packets, stations}, socket) do # Load the most recent packet for each RF path station station_packets = @@ -1532,6 +1537,9 @@ defmodule AprsmeWeb.MapLive.Index do Aprsme.SpatialPubSub.unregister_client(socket.assigns.spatial_client_id) end + # Unsubscribe from StreamingPacketsPubSub + Aprsme.StreamingPacketsPubSub.unsubscribe(self()) + if socket.assigns.buffer_timer, do: Process.cancel_timer(socket.assigns.buffer_timer) # Clean up any pending bounds update timer if socket.assigns[:bounds_update_timer] do @@ -1637,6 +1645,9 @@ defmodule AprsmeWeb.MapLive.Index do Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds) end + # Update StreamingPacketsPubSub subscription with new bounds + Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), map_bounds) + # Check if this is the initial load or if bounds have actually changed is_initial_load = socket.assigns[:needs_initial_historical_load] || !socket.assigns[:initial_bounds_loaded] bounds_changed = socket.assigns.map_bounds && not BoundsUtils.compare_bounds(map_bounds, socket.assigns.map_bounds)