graham/aprs.me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: reduce map live complexity

Browse source
This commit is contained in:
Graham McIntire 2026-03-22 17:20:27 -05:00
parent fba78766c8
commit 21d28afe8e
No known key found for this signature in database
3 changed files with 148 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
lib/aprsme/streaming_packets_pubsub.ex
View file

@ -112,26 +112,7 @@ defmodule Aprsme.StreamingPacketsPubSub do
@impl true
def handle_cast({:broadcast, packet}, state) do
# Get packet coordinates
lat = packet[:latitude] || packet[:lat]
lon = packet[:longitude] || packet[:lon] || packet[:lng]
if lat && lon do
# Get all subscribers — table is small (one entry per LiveView client)
subscribers = :ets.tab2list(@table_name)
if test_env?() do
send_to_matching_subscribers(subscribers, lat, lon, packet, self())
else
# Send to matching subscribers using BroadcastTaskSupervisor
# Collect dead pids to clean up
server_pid = self()
Aprsme.BroadcastTaskSupervisor.async_execute(fn ->
send_to_matching_subscribers(subscribers, lat, lon, packet, server_pid)
end)
end
end
maybe_broadcast_packet(packet)
{:noreply, state}
end
@ -216,4 +197,34 @@ defmodule Aprsme.StreamingPacketsPubSub do
default
end
end
defp maybe_broadcast_packet(packet) do
case packet_coordinates(packet) do
{lat, lon} ->
subscribers = :ets.tab2list(@table_name)
dispatch_packet_to_subscribers(subscribers, lat, lon, packet)
nil ->
:ok
end
end
defp packet_coordinates(packet) do
lat = packet[:latitude] || packet[:lat]
lon = packet[:longitude] || packet[:lon] || packet[:lng]
if lat && lon, do: {lat, lon}
end
defp dispatch_packet_to_subscribers(subscribers, lat, lon, packet) do
if test_env?() do
send_to_matching_subscribers(subscribers, lat, lon, packet, self())
else
server_pid = self()
Aprsme.BroadcastTaskSupervisor.async_execute(fn ->
send_to_matching_subscribers(subscribers, lat, lon, packet, server_pid)
end)
end
end
end

83
lib/aprsme_web/live/map_live/display_manager.ex
View file

@ -90,45 +90,12 @@ defmodule AprsmeWeb.MapLive.DisplayManager do
def send_trail_line_for_tracked_callsign(socket) do
callsign = socket.assigns.tracked_callsign
threshold = socket.assigns.packet_age_threshold
packets = trail_packets_for_callsign(socket.assigns.visible_packets, callsign, threshold)
packets =
socket.assigns.visible_packets
|> Map.values()
|> Enum.filter(fn packet ->
sender = Map.get(packet, :sender) || Map.get(packet, "sender")
received_at = Map.get(packet, :received_at) || Map.get(packet, "received_at")
String.upcase(sender) == String.upcase(callsign) &&
DateTime.compare(received_at, threshold) != :lt
end)
|> Enum.sort_by(
fn packet ->
received_at = Map.get(packet, :received_at) || Map.get(packet, "received_at")
DateTime.to_unix(received_at, :microsecond)
end,
:asc
)
if Enum.empty?(packets) do
if packets == [] do
socket
else
points =
Enum.map(packets, fn packet ->
lat = Map.get(packet, :lat) || Map.get(packet, "lat")
lon = Map.get(packet, :lon) || Map.get(packet, "lon")
received_at = Map.get(packet, :received_at) || Map.get(packet, "received_at")
%{
lat: Aprsme.EncodingUtils.to_float(lat) || 0.0,
lng: Aprsme.EncodingUtils.to_float(lon) || 0.0,
timestamp: DateTime.to_iso8601(received_at)
}
end)
LiveView.push_event(socket, "show_trail_line", %{
callsign: callsign,
points: points
})
push_trail_line(socket, callsign, packets)
end
end
@ -184,4 +151,48 @@ defmodule AprsmeWeb.MapLive.DisplayManager do
# This function should be moved to HistoricalLoader module
socket
end
defp trail_packets_for_callsign(visible_packets, callsign, threshold) do
visible_packets
|> Map.values()
|> Enum.filter(&tracked_packet?(&1, callsign, threshold))
|> Enum.sort_by(&packet_timestamp/1, :asc)
end
defp tracked_packet?(packet, callsign, threshold) do
sender = Map.get(packet, :sender) || Map.get(packet, "sender")
received_at = packet_received_at(packet)
String.upcase(sender) == String.upcase(callsign) &&
DateTime.compare(received_at, threshold) != :lt
end
defp push_trail_line(socket, callsign, packets) do
LiveView.push_event(socket, "show_trail_line", %{
callsign: callsign,
points: Enum.map(packets, &trail_point/1)
})
end
defp trail_point(packet) do
lat = Map.get(packet, :lat) || Map.get(packet, "lat")
lon = Map.get(packet, :lon) || Map.get(packet, "lon")
received_at = packet_received_at(packet)
%{
lat: Aprsme.EncodingUtils.to_float(lat) || 0.0,
lng: Aprsme.EncodingUtils.to_float(lon) || 0.0,
timestamp: DateTime.to_iso8601(received_at)
}
end
defp packet_received_at(packet) do
Map.get(packet, :received_at) || Map.get(packet, "received_at")
end
defp packet_timestamp(packet) do
packet
|> packet_received_at()
|> DateTime.to_unix(:microsecond)
end
end

117
lib/aprsme_web/live/map_live/index.ex
View file

@ -1945,60 +1945,22 @@ defmodule AprsmeWeb.MapLive.Index do
Logger.debug("process_bounds_update called with bounds: #{inspect(map_bounds)}")
# Update spatial viewport if we have a client ID
if socket.assigns[:spatial_client_id] do
Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds)
end
sync_pubsub_bounds(socket, map_bounds)
# Update StreamingPacketsPubSub subscription with new bounds
Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), map_bounds)
bounds_state = bounds_update_state(socket, map_bounds)
log_bounds_update_state(bounds_state)
# 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)
{new_visible_packets, packets_to_remove} = filtered_packets_for_bounds(socket.assigns.visible_packets, map_bounds)
# Check if we've completed the initial historical load
initial_historical_completed = socket.assigns[:initial_historical_completed] || false
Logger.debug(
"is_initial_load: #{is_initial_load}, bounds_changed: #{bounds_changed}, initial_historical_completed: #{initial_historical_completed}"
)
# Remove out-of-bounds packets and markers immediately
new_visible_packets = BoundsUtils.filter_packets_by_bounds(socket.assigns.visible_packets, map_bounds)
packets_to_remove = BoundsUtils.reject_packets_by_bounds(socket.assigns.visible_packets, map_bounds)
# Remove markers for out-of-bounds packets
socket = remove_markers_batch(socket, packets_to_remove)
# Only clear historical packets if:
# 1. Bounds actually changed AND
# 2. This is not the initial load AND
# 3. We've already completed the initial historical load
socket =
if bounds_changed and not is_initial_load and initial_historical_completed do
Logger.debug("Bounds changed after initial load - clearing historical packets")
push_event(socket, "clear_historical_packets", %{})
else
Logger.debug("Initial load or no significant change - keeping existing markers")
socket
end
# Always filter markers by bounds
socket = push_event(socket, "filter_markers_by_bounds", %{bounds: map_bounds})
# Update map bounds and visible_packets with bounds-filtered result
socket =
socket
|> remove_markers_batch(packets_to_remove)
|> maybe_clear_historical_packets(bounds_state)
|> push_event("filter_markers_by_bounds", %{bounds: map_bounds})
|> assign(map_bounds: map_bounds, visible_packets: new_visible_packets)
|> assign(needs_initial_historical_load: false)
# Load historical packets for the new bounds (now socket.assigns.map_bounds is correct)
Logger.debug("Starting progressive historical loading for new bounds")
Logger.debug(
"Current assigns: historical_loading=#{socket.assigns.historical_loading}, map_bounds=#{inspect(socket.assigns.map_bounds)}"
)
log_historical_reload(socket)
socket = HistoricalLoader.start_progressive_historical_loading(socket)
@ -2012,10 +1974,71 @@ defmodule AprsmeWeb.MapLive.Index do
end
# Mark initial historical as completed if this was the initial load
if is_initial_load do
if bounds_state.is_initial_load do
assign(socket, initial_historical_completed: true)
else
socket
end
end
defp sync_pubsub_bounds(socket, map_bounds) do
if socket.assigns[:spatial_client_id] do
Aprsme.SpatialPubSub.update_viewport(socket.assigns.spatial_client_id, map_bounds)
end
Aprsme.StreamingPacketsPubSub.subscribe_to_bounds(self(), map_bounds)
end
defp bounds_update_state(socket, map_bounds) do
%{
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),
initial_historical_completed: socket.assigns[:initial_historical_completed] || false
}
end
defp log_bounds_update_state(bounds_state) do
require Logger
Logger.debug(
"is_initial_load: #{bounds_state.is_initial_load}, bounds_changed: #{bounds_state.bounds_changed}, " <>
"initial_historical_completed: #{bounds_state.initial_historical_completed}"
)
end
defp filtered_packets_for_bounds(visible_packets, map_bounds) do
{
BoundsUtils.filter_packets_by_bounds(visible_packets, map_bounds),
BoundsUtils.reject_packets_by_bounds(visible_packets, map_bounds)
}
end
defp maybe_clear_historical_packets(socket, %{
bounds_changed: true,
is_initial_load: false,
initial_historical_completed: true
}) do
require Logger
Logger.debug("Bounds changed after initial load - clearing historical packets")
push_event(socket, "clear_historical_packets", %{})
end
defp maybe_clear_historical_packets(socket, _bounds_state) do
require Logger
Logger.debug("Initial load or no significant change - keeping existing markers")
socket
end
defp log_historical_reload(socket) do
require Logger
Logger.debug("Starting progressive historical loading for new bounds")
Logger.debug(
"Current assigns: historical_loading=#{socket.assigns.historical_loading}, map_bounds=#{inspect(socket.assigns.map_bounds)}"
)
end
end