Batch push_events, RF path queries, and ETS tuning for hot paths
- Batch real-time packet push_events: collect marker data across the packet batch and send a single "new_packets" WebSocket message instead of one per packet. JS handler processes all live→historical conversions before adding new markers to avoid intra-batch conflicts. - Batch RF path station queries: replace N+1 get_latest_packet_for_callsign calls with a single DISTINCT ON query selecting only callsign + lat/lng. - Add read_concurrency: true to streaming_packets_subscribers and :aprsme ETS tables (read-heavy, rarely written). - Remove dead GC code in PacketConsumer (threshold > 1000 unreachable with max batch_size of 100).
This commit is contained in:
parent
c377dfaf67
commit
dcabb744df
11 changed files with 462 additions and 64 deletions
|
|
@ -1125,6 +1125,89 @@ let MapAPRSMap = {
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Handle batched new packets from LiveView (performance optimization)
|
||||||
|
self.handleEvent(
|
||||||
|
"new_packets",
|
||||||
|
(payload: { packets: MarkerData[] }) => {
|
||||||
|
try {
|
||||||
|
if (!self || !self.map || self.isDestroyed) return;
|
||||||
|
if (!self.map.hasLayer) return;
|
||||||
|
if (self.heatLayer && self.map.hasLayer(self.heatLayer)) return;
|
||||||
|
if (!self.markerStates || !self.markers) return;
|
||||||
|
|
||||||
|
const packets = payload.packets;
|
||||||
|
if (!packets || packets.length === 0) return;
|
||||||
|
|
||||||
|
// Phase 1: Collect ALL markers to convert across the entire batch
|
||||||
|
// This prevents converting a marker that was just added by an earlier packet
|
||||||
|
const allNewCallsigns = new Set<string>();
|
||||||
|
const allNewIds = new Set<string>();
|
||||||
|
for (const data of packets) {
|
||||||
|
const cs = data.callsign_group || data.callsign || data.id;
|
||||||
|
if (cs) allNewCallsigns.add(cs);
|
||||||
|
if (data.id) allNewIds.add(String(data.id));
|
||||||
|
}
|
||||||
|
|
||||||
|
const markersToConvert: string[] = [];
|
||||||
|
self.markerStates.forEach((state, id) => {
|
||||||
|
const stateCallsign = state.callsign_group || state.callsign;
|
||||||
|
if (
|
||||||
|
stateCallsign &&
|
||||||
|
allNewCallsigns.has(stateCallsign) &&
|
||||||
|
(!state.historical || state.is_most_recent_for_callsign) &&
|
||||||
|
!allNewIds.has(String(id))
|
||||||
|
) {
|
||||||
|
markersToConvert.push(String(id));
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
// Phase 2: Convert all existing live markers to historical dots
|
||||||
|
for (const id of markersToConvert) {
|
||||||
|
const existingMarker = self.markers.get(id);
|
||||||
|
const existingState = self.markerStates.get(id);
|
||||||
|
if (existingMarker && existingState) {
|
||||||
|
existingState.historical = true;
|
||||||
|
existingState.is_most_recent_for_callsign = false;
|
||||||
|
const incomingCallsign =
|
||||||
|
existingState.callsign_group || existingState.callsign;
|
||||||
|
const historicalIconData = {
|
||||||
|
id: String(id),
|
||||||
|
lat: existingState.lat,
|
||||||
|
lng: existingState.lng,
|
||||||
|
callsign: existingState.callsign || incomingCallsign,
|
||||||
|
callsign_group:
|
||||||
|
existingState.callsign_group || incomingCallsign,
|
||||||
|
symbol_table_id: existingState.symbol_table,
|
||||||
|
symbol_code: existingState.symbol_code,
|
||||||
|
historical: true,
|
||||||
|
is_most_recent_for_callsign: false,
|
||||||
|
popup: existingState.popup,
|
||||||
|
};
|
||||||
|
existingMarker.setIcon(self.createMarkerIcon(historicalIconData));
|
||||||
|
(existingMarker as APRSMarker)._isHistorical = true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Phase 3: Add all new markers
|
||||||
|
for (const data of packets) {
|
||||||
|
const shouldOpenPopup = data.openPopup !== false;
|
||||||
|
const incomingCallsign =
|
||||||
|
data.callsign_group || data.callsign || data.id;
|
||||||
|
self.addMarker({
|
||||||
|
...data,
|
||||||
|
historical: false,
|
||||||
|
is_most_recent_for_callsign: true,
|
||||||
|
callsign_group: incomingCallsign,
|
||||||
|
popup: data.popup || self.buildPopupContent(data),
|
||||||
|
openPopup: shouldOpenPopup,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
console.error("Error in new_packets handler:", error);
|
||||||
|
}
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
// Handle highlighting the latest packet (open its popup)
|
// Handle highlighting the latest packet (open its popup)
|
||||||
self.currentPopupMarkerId = null;
|
self.currentPopupMarkerId = null;
|
||||||
self.handleEvent("highlight_packet", (data: { id: string }) => {
|
self.handleEvent("highlight_packet", (data: { id: string }) => {
|
||||||
|
|
|
||||||
|
|
@ -185,7 +185,7 @@ defmodule Aprsme.Application do
|
||||||
:ets.new(:query_cache, [:set, :public, :named_table, read_concurrency: true])
|
:ets.new(:query_cache, [:set, :public, :named_table, read_concurrency: true])
|
||||||
:ets.new(:device_cache, [:set, :public, :named_table, read_concurrency: true])
|
:ets.new(:device_cache, [:set, :public, :named_table, read_concurrency: true])
|
||||||
:ets.new(:symbol_cache, [:set, :public, :named_table, read_concurrency: true])
|
:ets.new(:symbol_cache, [:set, :public, :named_table, read_concurrency: true])
|
||||||
:ets.new(:aprsme, [:set, :public, :named_table, write_concurrency: true])
|
:ets.new(:aprsme, [:set, :public, :named_table, read_concurrency: true])
|
||||||
:ets.insert(:aprsme, {:message_number, 0})
|
:ets.insert(:aprsme, {:message_number, 0})
|
||||||
|
|
||||||
[
|
[
|
||||||
|
|
|
||||||
|
|
@ -232,12 +232,6 @@ defmodule Aprsme.PacketConsumer do
|
||||||
)
|
)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Only do minor GC after very large batches to prevent memory accumulation
|
|
||||||
# This should rarely trigger with batch_size of 100
|
|
||||||
if length(packets) > 1000 do
|
|
||||||
:erlang.garbage_collect(self(), type: :minor)
|
|
||||||
end
|
|
||||||
|
|
||||||
:telemetry.execute(
|
:telemetry.execute(
|
||||||
[
|
[
|
||||||
:aprsme,
|
:aprsme,
|
||||||
|
|
|
||||||
|
|
@ -866,6 +866,15 @@ defmodule Aprsme.Packets do
|
||||||
PreparedQueries.get_latest_packet_for_callsign(callsign)
|
PreparedQueries.get_latest_packet_for_callsign(callsign)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Gets latest positions for multiple callsigns in a single batch query.
|
||||||
|
Returns a list of `%{callsign: String.t(), lat: float(), lng: float()}` maps.
|
||||||
|
"""
|
||||||
|
@spec get_latest_positions_for_callsigns(list(String.t())) :: [map()]
|
||||||
|
def get_latest_positions_for_callsigns(callsigns) when is_list(callsigns) do
|
||||||
|
PreparedQueries.get_latest_positions_for_callsigns(callsigns)
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Gets the most recent weather packet for a callsign.
|
Gets the most recent weather packet for a callsign.
|
||||||
Looks for any packet containing weather data fields.
|
Looks for any packet containing weather data fields.
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,28 @@ defmodule Aprsme.Packets.PreparedQueries do
|
||||||
)
|
)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Get latest positions for multiple callsigns in a single query.
|
||||||
|
Returns a list of `%{callsign: String.t(), lat: float(), lng: float()}` maps.
|
||||||
|
Uses DISTINCT ON to get only the most recent packet per callsign.
|
||||||
|
"""
|
||||||
|
@spec get_latest_positions_for_callsigns(list(String.t())) :: [map()]
|
||||||
|
def get_latest_positions_for_callsigns([]), do: []
|
||||||
|
|
||||||
|
def get_latest_positions_for_callsigns(callsigns) when is_list(callsigns) do
|
||||||
|
normalized = Enum.map(callsigns, &String.upcase(String.trim(&1)))
|
||||||
|
|
||||||
|
Repo.all(
|
||||||
|
from(p in Packet,
|
||||||
|
where: fragment("upper(?)", p.sender) in ^normalized,
|
||||||
|
where: not is_nil(p.location),
|
||||||
|
distinct: fragment("upper(?)", p.sender),
|
||||||
|
order_by: [asc: fragment("upper(?)", p.sender), desc: p.received_at],
|
||||||
|
select: %{callsign: p.sender, lat: fragment("ST_Y(?)", p.location), lng: fragment("ST_X(?)", p.location)}
|
||||||
|
)
|
||||||
|
)
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Get recent packets within bounds using prepared statement.
|
Get recent packets within bounds using prepared statement.
|
||||||
"""
|
"""
|
||||||
|
|
|
||||||
|
|
@ -61,7 +61,7 @@ defmodule Aprsme.StreamingPacketsPubSub do
|
||||||
@impl true
|
@impl true
|
||||||
def init(_opts) do
|
def init(_opts) do
|
||||||
# Create ETS table for fast lookups
|
# Create ETS table for fast lookups
|
||||||
:ets.new(@table_name, [:set, :protected, :named_table])
|
:ets.new(@table_name, [:set, :protected, :named_table, read_concurrency: true])
|
||||||
|
|
||||||
# Monitor subscribers for cleanup
|
# Monitor subscribers for cleanup
|
||||||
Process.flag(:trap_exit, true)
|
Process.flag(:trap_exit, true)
|
||||||
|
|
|
||||||
|
|
@ -850,15 +850,38 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
end
|
end
|
||||||
|
|
||||||
def handle_info({:packet_batch, packets}, socket) do
|
def handle_info({:packet_batch, packets}, socket) do
|
||||||
# Process batch of packets efficiently
|
# Process batch: collect marker data without pushing events, then send one batch push
|
||||||
socket =
|
{socket, marker_data_list} =
|
||||||
Enum.reduce(packets, socket, fn packet, acc_socket ->
|
Enum.reduce(packets, {socket, []}, fn packet, {acc_socket, markers} ->
|
||||||
case handle_info_postgres_packet(packet, acc_socket) do
|
{new_socket, marker_data} = process_packet_for_batch(packet, acc_socket)
|
||||||
{:noreply, new_socket} -> new_socket
|
markers = if marker_data, do: [marker_data | markers], else: markers
|
||||||
_ -> acc_socket
|
{new_socket, markers}
|
||||||
end
|
|
||||||
end)
|
end)
|
||||||
|
|
||||||
|
socket = assign(socket, :last_update_at, DateTime.utc_now())
|
||||||
|
|
||||||
|
# Send heat map update once if in heat map mode, otherwise batch markers
|
||||||
|
socket =
|
||||||
|
cond do
|
||||||
|
socket.assigns.map_zoom <= 8 and marker_data_list != [] ->
|
||||||
|
DisplayManager.send_heat_map_for_current_bounds(socket)
|
||||||
|
|
||||||
|
marker_data_list != [] ->
|
||||||
|
markers =
|
||||||
|
Enum.map(Enum.reverse(marker_data_list), fn data ->
|
||||||
|
if socket.assigns.station_popup_open do
|
||||||
|
Map.put(data, :openPopup, false)
|
||||||
|
else
|
||||||
|
data
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
push_event(socket, "new_packets", %{packets: markers})
|
||||||
|
|
||||||
|
true ->
|
||||||
|
socket
|
||||||
|
end
|
||||||
|
|
||||||
{:noreply, socket}
|
{:noreply, socket}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
@ -1047,6 +1070,21 @@ defmodule AprsmeWeb.MapLive.Index do
|
||||||
PacketProcessor.process_packet_for_display(packet, socket)
|
PacketProcessor.process_packet_for_display(packet, socket)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
defp process_packet_for_batch(packet, socket) do
|
||||||
|
if socket.assigns.tracked_callsign == "" do
|
||||||
|
PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
else
|
||||||
|
packet_sender = Map.get(packet, :sender, Map.get(packet, "sender", ""))
|
||||||
|
|
||||||
|
if String.upcase(packet_sender) == String.upcase(socket.assigns.tracked_callsign) do
|
||||||
|
socket = assign(socket, :tracked_callsign_latest_packet, packet)
|
||||||
|
PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
else
|
||||||
|
{socket, nil}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
# Handle replaying the next historical packet
|
# Handle replaying the next historical packet
|
||||||
|
|
||||||
@impl true
|
@impl true
|
||||||
|
|
|
||||||
|
|
@ -9,8 +9,8 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
alias AprsmeWeb.Live.Shared.BoundsUtils
|
alias AprsmeWeb.Live.Shared.BoundsUtils
|
||||||
alias AprsmeWeb.Live.Shared.CoordinateUtils
|
alias AprsmeWeb.Live.Shared.CoordinateUtils
|
||||||
alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils
|
alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils
|
||||||
|
alias AprsmeWeb.MapLive.DataBuilder
|
||||||
alias AprsmeWeb.MapLive.DisplayManager
|
alias AprsmeWeb.MapLive.DisplayManager
|
||||||
alias AprsmeWeb.MapLive.PacketUtils
|
|
||||||
alias Phoenix.LiveView
|
alias Phoenix.LiveView
|
||||||
alias Phoenix.LiveView.Socket
|
alias Phoenix.LiveView.Socket
|
||||||
|
|
||||||
|
|
@ -18,6 +18,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Process a packet for display on the map.
|
Process a packet for display on the map.
|
||||||
|
Pushes events directly to the socket (used for single-packet path).
|
||||||
"""
|
"""
|
||||||
@spec process_packet_for_display(map(), Socket.t()) :: {:noreply, Socket.t()}
|
@spec process_packet_for_display(map(), Socket.t()) :: {:noreply, Socket.t()}
|
||||||
def process_packet_for_display(packet, socket) do
|
def process_packet_for_display(packet, socket) do
|
||||||
|
|
@ -28,12 +29,24 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
socket = handle_packet_visibility(packet, lat, lon, callsign_key, socket)
|
socket = handle_packet_visibility(packet, lat, lon, callsign_key, socket)
|
||||||
|
|
||||||
# Update last update timestamp for real-time display in map sidebar
|
# Update last update timestamp for real-time display in map sidebar
|
||||||
# This timestamp is shown to users to indicate when data was last refreshed
|
|
||||||
socket = assign(socket, :last_update_at, DateTime.utc_now())
|
socket = assign(socket, :last_update_at, DateTime.utc_now())
|
||||||
|
|
||||||
{:noreply, socket}
|
{:noreply, socket}
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@doc """
|
||||||
|
Process a packet for batch display. Updates socket state but does NOT push
|
||||||
|
events. Returns `{socket, marker_data | nil}` where marker_data is the data
|
||||||
|
that should be included in a batched push_event, or nil if no marker needed.
|
||||||
|
"""
|
||||||
|
@spec process_packet_for_marker_data(map(), Socket.t()) :: {Socket.t(), map() | nil}
|
||||||
|
def process_packet_for_marker_data(packet, socket) do
|
||||||
|
{lat, lon, _data_extended} = CoordinateUtils.get_coordinates(packet)
|
||||||
|
callsign_key = SharedPacketUtils.get_callsign_key(packet)
|
||||||
|
|
||||||
|
batch_visibility_action(packet, lat, lon, callsign_key, socket)
|
||||||
|
end
|
||||||
|
|
||||||
@doc """
|
@doc """
|
||||||
Handle packet visibility logic based on bounds and existing markers.
|
Handle packet visibility logic based on bounds and existing markers.
|
||||||
"""
|
"""
|
||||||
|
|
@ -78,34 +91,17 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
"""
|
"""
|
||||||
@spec handle_valid_postgres_packet(map(), number() | nil, number() | nil, Socket.t()) :: Socket.t()
|
@spec handle_valid_postgres_packet(map(), number() | nil, number() | nil, Socket.t()) :: Socket.t()
|
||||||
def handle_valid_postgres_packet(packet, _lat, _lon, socket) do
|
def handle_valid_postgres_packet(packet, _lat, _lon, socket) do
|
||||||
# Add the packet to visible_packets and push a marker immediately
|
{socket, marker_data} = prepare_packet_state(packet, socket)
|
||||||
callsign_key = SharedPacketUtils.get_callsign_key(packet)
|
|
||||||
|
|
||||||
new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet)
|
cond do
|
||||||
|
socket.assigns.map_zoom <= 8 ->
|
||||||
|
send_heat_map_for_current_bounds(socket)
|
||||||
|
|
||||||
# Enforce memory limits
|
marker_data ->
|
||||||
new_visible_packets =
|
|
||||||
if map_size(new_visible_packets) > @max_visible_packets do
|
|
||||||
SharedPacketUtils.prune_oldest_packets(new_visible_packets, @max_visible_packets)
|
|
||||||
else
|
|
||||||
new_visible_packets
|
|
||||||
end
|
|
||||||
|
|
||||||
socket = assign(socket, :visible_packets, new_visible_packets)
|
|
||||||
|
|
||||||
# Check zoom level to decide how to display the packet
|
|
||||||
if socket.assigns.map_zoom <= 8 do
|
|
||||||
# We're in heat map mode - update the heat map with all current data
|
|
||||||
send_heat_map_for_current_bounds(socket)
|
|
||||||
else
|
|
||||||
# We're in marker mode - send individual marker
|
|
||||||
marker_data = PacketUtils.build_packet_data(packet, true, get_locale(socket))
|
|
||||||
|
|
||||||
if marker_data do
|
|
||||||
send_marker_with_popup_check(socket, marker_data)
|
send_marker_with_popup_check(socket, marker_data)
|
||||||
else
|
|
||||||
|
true ->
|
||||||
socket
|
socket
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
@ -119,14 +115,78 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
assign(socket, :visible_packets, new_visible_packets)
|
assign(socket, :visible_packets, new_visible_packets)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
# Batch-mode visibility: same logic as visibility_action but returns
|
||||||
|
# {socket, marker_data | nil} without pushing events
|
||||||
|
defp batch_visibility_action(packet, lat, lon, callsign_key, socket) when not is_nil(lat) and not is_nil(lon) do
|
||||||
|
in_bounds = within_bounds?(%{lat: lat, lon: lon}, socket.assigns.map_bounds)
|
||||||
|
has_marker = Map.has_key?(socket.assigns.visible_packets, callsign_key)
|
||||||
|
batch_dispatch({in_bounds, has_marker}, packet, lat, lon, callsign_key, socket)
|
||||||
|
end
|
||||||
|
|
||||||
|
defp batch_visibility_action(_packet, _lat, _lon, _callsign_key, socket) do
|
||||||
|
{socket, nil}
|
||||||
|
end
|
||||||
|
|
||||||
|
# Out of bounds but has marker — remove it (still pushes remove_marker since
|
||||||
|
# removals are rare and don't benefit from batching)
|
||||||
|
defp batch_dispatch({false, true}, _packet, _lat, _lon, callsign_key, socket) do
|
||||||
|
socket = remove_marker_from_map(callsign_key, socket)
|
||||||
|
{socket, nil}
|
||||||
|
end
|
||||||
|
|
||||||
|
# In bounds, no existing marker — add it
|
||||||
|
defp batch_dispatch({true, false}, packet, _lat, _lon, _callsign_key, socket) do
|
||||||
|
prepare_packet_state(packet, socket)
|
||||||
|
end
|
||||||
|
|
||||||
|
# In bounds, existing marker — update if significant movement
|
||||||
|
defp batch_dispatch({true, true}, packet, lat, lon, callsign_key, socket) do
|
||||||
|
existing_packet = socket.assigns.visible_packets[callsign_key]
|
||||||
|
{existing_lat, existing_lon, _} = CoordinateUtils.get_coordinates(existing_packet)
|
||||||
|
|
||||||
|
if is_number(existing_lat) and is_number(existing_lon) and
|
||||||
|
GeoUtils.significant_movement?(existing_lat, existing_lon, lat, lon, 50) do
|
||||||
|
prepare_packet_state(packet, socket)
|
||||||
|
else
|
||||||
|
new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet)
|
||||||
|
{assign(socket, :visible_packets, new_visible_packets), nil}
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
defp batch_dispatch(_, _packet, _lat, _lon, _callsign_key, socket) do
|
||||||
|
{socket, nil}
|
||||||
|
end
|
||||||
|
|
||||||
|
# Shared: updates visible_packets state and builds marker data.
|
||||||
|
# Returns {socket, marker_data | nil}. Does NOT push events.
|
||||||
|
defp prepare_packet_state(packet, socket) do
|
||||||
|
callsign_key = SharedPacketUtils.get_callsign_key(packet)
|
||||||
|
|
||||||
|
new_visible_packets = Map.put(socket.assigns.visible_packets, callsign_key, packet)
|
||||||
|
|
||||||
|
new_visible_packets =
|
||||||
|
if map_size(new_visible_packets) > @max_visible_packets do
|
||||||
|
SharedPacketUtils.prune_oldest_packets(new_visible_packets, @max_visible_packets)
|
||||||
|
else
|
||||||
|
new_visible_packets
|
||||||
|
end
|
||||||
|
|
||||||
|
socket = assign(socket, :visible_packets, new_visible_packets)
|
||||||
|
|
||||||
|
marker_data =
|
||||||
|
if socket.assigns.map_zoom > 8 do
|
||||||
|
DataBuilder.build_packet_data(packet, true, get_locale(socket))
|
||||||
|
end
|
||||||
|
|
||||||
|
{socket, marker_data}
|
||||||
|
end
|
||||||
|
|
||||||
# Private helper functions
|
# Private helper functions
|
||||||
|
|
||||||
# Use shared bounds utility
|
|
||||||
defp within_bounds?(coords, bounds) do
|
defp within_bounds?(coords, bounds) do
|
||||||
BoundsUtils.within_bounds?(coords, bounds)
|
BoundsUtils.within_bounds?(coords, bounds)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Helper to get locale from socket
|
|
||||||
defp get_locale(socket) do
|
defp get_locale(socket) do
|
||||||
Map.get(socket.assigns, :locale, "en")
|
Map.get(socket.assigns, :locale, "en")
|
||||||
end
|
end
|
||||||
|
|
@ -137,7 +197,6 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
||||||
|
|
||||||
defp send_marker_with_popup_check(socket, marker_data) do
|
defp send_marker_with_popup_check(socket, marker_data) do
|
||||||
if socket.assigns.station_popup_open do
|
if socket.assigns.station_popup_open do
|
||||||
# Send without opening popup to avoid interrupting user
|
|
||||||
LiveView.push_event(socket, "new_packet", Map.put(marker_data, :openPopup, false))
|
LiveView.push_event(socket, "new_packet", Map.put(marker_data, :openPopup, false))
|
||||||
else
|
else
|
||||||
LiveView.push_event(socket, "new_packet", marker_data)
|
LiveView.push_event(socket, "new_packet", marker_data)
|
||||||
|
|
|
||||||
|
|
@ -46,12 +46,9 @@ defmodule AprsmeWeb.MapLive.RfPath do
|
||||||
"""
|
"""
|
||||||
@spec get_path_station_positions(list(binary()), Phoenix.LiveView.Socket.t()) :: list(map())
|
@spec get_path_station_positions(list(binary()), Phoenix.LiveView.Socket.t()) :: list(map())
|
||||||
def get_path_station_positions(path_stations, _socket) do
|
def get_path_station_positions(path_stations, _socket) do
|
||||||
# Limit to prevent excessive database queries
|
path_stations
|
||||||
limited_stations = Enum.take(path_stations, 10)
|
|> Enum.take(10)
|
||||||
|
|> Packets.get_latest_positions_for_callsigns()
|
||||||
limited_stations
|
|
||||||
|> Enum.map(&get_station_position/1)
|
|
||||||
|> Enum.filter(& &1)
|
|
||||||
end
|
end
|
||||||
|
|
||||||
defp split_at_q_construct(elements) do
|
defp split_at_q_construct(elements) do
|
||||||
|
|
@ -95,18 +92,4 @@ defmodule AprsmeWeb.MapLive.RfPath do
|
||||||
""
|
""
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
defp get_station_position(callsign) do
|
|
||||||
case Packets.get_latest_packet_for_callsign(callsign) do
|
|
||||||
%{lat: lat, lon: lon} when is_number(lat) and is_number(lon) ->
|
|
||||||
%{
|
|
||||||
callsign: callsign,
|
|
||||||
lat: lat,
|
|
||||||
lng: lon
|
|
||||||
}
|
|
||||||
|
|
||||||
_ ->
|
|
||||||
nil
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
|
|
|
||||||
113
test/aprsme/packets/prepared_queries_test.exs
Normal file
113
test/aprsme/packets/prepared_queries_test.exs
Normal file
|
|
@ -0,0 +1,113 @@
|
||||||
|
defmodule Aprsme.Packets.PreparedQueriesTest do
|
||||||
|
use Aprsme.DataCase, async: false
|
||||||
|
|
||||||
|
alias Aprsme.Packet
|
||||||
|
alias Aprsme.Packets.PreparedQueries
|
||||||
|
alias Aprsme.Repo
|
||||||
|
|
||||||
|
defp create_positioned_packet(attrs) do
|
||||||
|
defaults = %{
|
||||||
|
sender: "TEST-1",
|
||||||
|
base_callsign: "TEST",
|
||||||
|
ssid: "1",
|
||||||
|
destination: "APRS",
|
||||||
|
data_type: "position",
|
||||||
|
lat: Decimal.new("33.0000"),
|
||||||
|
lon: Decimal.new("-96.0000"),
|
||||||
|
received_at: DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
|
}
|
||||||
|
|
||||||
|
merged = Map.merge(defaults, attrs)
|
||||||
|
|
||||||
|
%Packet{}
|
||||||
|
|> Packet.changeset(Map.from_struct(Map.merge(%Packet{}, merged)))
|
||||||
|
|> Repo.insert()
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "get_latest_positions_for_callsigns/1" do
|
||||||
|
test "returns positions for multiple callsigns" do
|
||||||
|
{:ok, _} =
|
||||||
|
create_positioned_packet(%{
|
||||||
|
sender: "K5GVL-10",
|
||||||
|
base_callsign: "K5GVL",
|
||||||
|
ssid: "10",
|
||||||
|
lat: Decimal.new("33.1000"),
|
||||||
|
lon: Decimal.new("-96.6000")
|
||||||
|
})
|
||||||
|
|
||||||
|
{:ok, _} =
|
||||||
|
create_positioned_packet(%{
|
||||||
|
sender: "N5TXZ-10",
|
||||||
|
base_callsign: "N5TXZ",
|
||||||
|
ssid: "10",
|
||||||
|
lat: Decimal.new("33.2000"),
|
||||||
|
lon: Decimal.new("-96.5000")
|
||||||
|
})
|
||||||
|
|
||||||
|
result = PreparedQueries.get_latest_positions_for_callsigns(["K5GVL-10", "N5TXZ-10"])
|
||||||
|
|
||||||
|
assert length(result) == 2
|
||||||
|
callsigns = Enum.map(result, & &1.callsign)
|
||||||
|
assert "K5GVL-10" in callsigns
|
||||||
|
assert "N5TXZ-10" in callsigns
|
||||||
|
|
||||||
|
k5gvl = Enum.find(result, &(&1.callsign == "K5GVL-10"))
|
||||||
|
assert_in_delta k5gvl.lat, 33.1, 0.01
|
||||||
|
assert_in_delta k5gvl.lng, -96.6, 0.01
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns empty list for empty input" do
|
||||||
|
assert PreparedQueries.get_latest_positions_for_callsigns([]) == []
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns empty list for nonexistent callsigns" do
|
||||||
|
result = PreparedQueries.get_latest_positions_for_callsigns(["NONEXIST-1", "FAKE-2"])
|
||||||
|
assert result == []
|
||||||
|
end
|
||||||
|
|
||||||
|
test "is case-insensitive" do
|
||||||
|
{:ok, _} =
|
||||||
|
create_positioned_packet(%{
|
||||||
|
sender: "K5GVL-10",
|
||||||
|
base_callsign: "K5GVL",
|
||||||
|
ssid: "10",
|
||||||
|
lat: Decimal.new("33.1000"),
|
||||||
|
lon: Decimal.new("-96.6000")
|
||||||
|
})
|
||||||
|
|
||||||
|
result = PreparedQueries.get_latest_positions_for_callsigns(["k5gvl-10"])
|
||||||
|
|
||||||
|
assert length(result) == 1
|
||||||
|
assert hd(result).callsign == "K5GVL-10"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns latest position when multiple packets exist for a callsign" do
|
||||||
|
{:ok, _} =
|
||||||
|
create_positioned_packet(%{
|
||||||
|
sender: "K5GVL-10",
|
||||||
|
base_callsign: "K5GVL",
|
||||||
|
ssid: "10",
|
||||||
|
lat: Decimal.new("33.1000"),
|
||||||
|
lon: Decimal.new("-96.6000"),
|
||||||
|
received_at: DateTime.add(DateTime.utc_now(), -3600, :second)
|
||||||
|
})
|
||||||
|
|
||||||
|
{:ok, _} =
|
||||||
|
create_positioned_packet(%{
|
||||||
|
sender: "K5GVL-10",
|
||||||
|
base_callsign: "K5GVL",
|
||||||
|
ssid: "10",
|
||||||
|
lat: Decimal.new("34.0000"),
|
||||||
|
lon: Decimal.new("-97.0000"),
|
||||||
|
received_at: DateTime.truncate(DateTime.utc_now(), :second)
|
||||||
|
})
|
||||||
|
|
||||||
|
result = PreparedQueries.get_latest_positions_for_callsigns(["K5GVL-10"])
|
||||||
|
|
||||||
|
assert length(result) == 1
|
||||||
|
position = hd(result)
|
||||||
|
assert_in_delta position.lat, 34.0, 0.01
|
||||||
|
assert_in_delta position.lng, -97.0, 0.01
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
97
test/aprsme_web/live/map_live/packet_processor_test.exs
Normal file
97
test/aprsme_web/live/map_live/packet_processor_test.exs
Normal file
|
|
@ -0,0 +1,97 @@
|
||||||
|
defmodule AprsmeWeb.MapLive.PacketProcessorTest do
|
||||||
|
use ExUnit.Case, async: true
|
||||||
|
|
||||||
|
alias AprsmeWeb.MapLive.PacketProcessor
|
||||||
|
|
||||||
|
defp build_socket(assigns \\ %{}) do
|
||||||
|
defaults = %{
|
||||||
|
map_bounds: %{north: 34.0, south: 32.0, east: -95.0, west: -97.0},
|
||||||
|
visible_packets: %{},
|
||||||
|
map_zoom: 10,
|
||||||
|
station_popup_open: false,
|
||||||
|
locale: "en"
|
||||||
|
}
|
||||||
|
|
||||||
|
merged = Map.merge(defaults, assigns)
|
||||||
|
|
||||||
|
socket = %Phoenix.LiveView.Socket{
|
||||||
|
assigns: Map.put(merged, :__changed__, %{})
|
||||||
|
}
|
||||||
|
|
||||||
|
socket
|
||||||
|
end
|
||||||
|
|
||||||
|
defp build_packet(overrides \\ %{}) do
|
||||||
|
Map.merge(
|
||||||
|
%{
|
||||||
|
id: "test-packet-1",
|
||||||
|
sender: "K5GVL-10",
|
||||||
|
base_callsign: "K5GVL",
|
||||||
|
ssid: "10",
|
||||||
|
lat: 33.1,
|
||||||
|
lon: -96.5,
|
||||||
|
symbol_table_id: "/",
|
||||||
|
symbol_code: "-",
|
||||||
|
comment: "Test station",
|
||||||
|
received_at: DateTime.utc_now()
|
||||||
|
},
|
||||||
|
overrides
|
||||||
|
)
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "process_packet_for_marker_data/2" do
|
||||||
|
test "returns marker data for in-bounds packet at marker zoom level" do
|
||||||
|
socket = build_socket()
|
||||||
|
packet = build_packet()
|
||||||
|
|
||||||
|
{updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
|
||||||
|
assert marker_data
|
||||||
|
assert is_map(marker_data)
|
||||||
|
# Socket state should be updated
|
||||||
|
assert map_size(updated_socket.assigns.visible_packets) == 1
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns nil marker data for out-of-bounds packet" do
|
||||||
|
socket = build_socket()
|
||||||
|
# Packet outside the bounds (north: 34, south: 32, east: -95, west: -97)
|
||||||
|
packet = build_packet(%{lat: 40.0, lon: -80.0})
|
||||||
|
|
||||||
|
{updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
|
||||||
|
assert marker_data == nil
|
||||||
|
assert map_size(updated_socket.assigns.visible_packets) == 0
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns nil marker data for packet with nil coordinates" do
|
||||||
|
socket = build_socket()
|
||||||
|
packet = build_packet(%{lat: nil, lon: nil})
|
||||||
|
|
||||||
|
{updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
|
||||||
|
assert marker_data == nil
|
||||||
|
assert map_size(updated_socket.assigns.visible_packets) == 0
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns nil marker data in heat map mode (zoom <= 8)" do
|
||||||
|
socket = build_socket(%{map_zoom: 7})
|
||||||
|
packet = build_packet()
|
||||||
|
|
||||||
|
{_updated_socket, marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
|
||||||
|
assert marker_data == nil
|
||||||
|
end
|
||||||
|
|
||||||
|
test "does not push any events to socket" do
|
||||||
|
socket = build_socket()
|
||||||
|
packet = build_packet()
|
||||||
|
|
||||||
|
{updated_socket, _marker_data} = PacketProcessor.process_packet_for_marker_data(packet, socket)
|
||||||
|
|
||||||
|
# Socket should not have any push_events queued
|
||||||
|
# In LiveView, push_event adds to socket.private[:push_events]
|
||||||
|
push_events = get_in(updated_socket.private, [:push_events]) || []
|
||||||
|
assert push_events == []
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
Loading…
Add table
Reference in a new issue