Fix code review findings batch 2: dead code, encoding, correctness
- Delete dead modules: Archiver, PacketPipelineSetup, SystemMonitor - Encoding: fast-path for clean ASCII, fix latin1 C1 control chars (128-159) - SpatialPubSub: fix date line grid cell wrapping - QueryBuilder: use indexed has_weather column instead of 10-way OR - PacketProcessor/HistoricalLoader: wire up heat map via DisplayManager - MapLive: add PacketBatcher crash recovery with Process.monitor - MapLive: fix time display re-render by touching assign - DeviceCache: remove unreachable outer try/rescue - Remove dead stubs: RateLimiterWrapper count/reset, DeviceIdentification enqueue_refresh_job - Simplify Application.pubsub_config/0 - Add tests for SpatialPubSub, PacketBatcher, encoding, weather_only
This commit is contained in:
parent
b5eb59d169
commit
7b488e9fd7
18 changed files with 248 additions and 828 deletions
|
|
@ -177,22 +177,7 @@ defmodule Aprsme.Application do
|
|||
end
|
||||
|
||||
defp pubsub_config do
|
||||
cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
|
||||
|
||||
if cluster_enabled do
|
||||
require Logger
|
||||
|
||||
Logger.info("Starting distributed PubSub for clustering")
|
||||
|
||||
# Phoenix PubSub automatically uses distributed Erlang clustering when nodes are connected
|
||||
{Phoenix.PubSub, name: Aprsme.PubSub}
|
||||
else
|
||||
require Logger
|
||||
|
||||
Logger.info("Starting local PubSub adapter")
|
||||
|
||||
{Phoenix.PubSub, name: Aprsme.PubSub}
|
||||
end
|
||||
{Phoenix.PubSub, name: Aprsme.PubSub}
|
||||
end
|
||||
|
||||
# Removed - Exq configuration is now in runtime.exs
|
||||
|
|
|
|||
|
|
@ -1,35 +0,0 @@
|
|||
defmodule Aprsme.Archiver do
|
||||
@moduledoc false
|
||||
use GenServer
|
||||
|
||||
alias AprsmeWeb.Endpoint
|
||||
|
||||
require Jason
|
||||
require Logger
|
||||
|
||||
# alias Aprsme.{Packet, Repo}
|
||||
@topic "call"
|
||||
|
||||
# API
|
||||
@spec start_link(any) :: :ignore | {:error, any} | {:ok, pid}
|
||||
def start_link(_args \\ []) do
|
||||
GenServer.start_link(__MODULE__, [], name: :archiver)
|
||||
end
|
||||
|
||||
# Callbacks
|
||||
|
||||
@spec init(any) :: {:ok, any}
|
||||
def init(state \\ []) do
|
||||
Process.send_after(self(), :connect, 5000)
|
||||
Endpoint.subscribe(@topic)
|
||||
{:ok, state}
|
||||
end
|
||||
|
||||
def handle_info(:connect, state) do
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
def handle_info(_msg, state) do
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
|
|
@ -73,18 +73,11 @@ defmodule Aprsme.DeviceCache do
|
|||
|
||||
@impl true
|
||||
def handle_info(:initial_load, state) do
|
||||
# Load devices on startup
|
||||
case load_devices_into_cache() do
|
||||
:ok ->
|
||||
# Schedule periodic refresh
|
||||
Process.send_after(self(), :refresh_cache, @refresh_interval)
|
||||
{:noreply, %{state | initial_load_done: true}}
|
||||
load_devices_into_cache()
|
||||
|
||||
:error ->
|
||||
# If initial load failed, retry sooner
|
||||
Process.send_after(self(), :initial_load, 5_000)
|
||||
{:noreply, state}
|
||||
end
|
||||
# Schedule periodic refresh
|
||||
Process.send_after(self(), :refresh_cache, @refresh_interval)
|
||||
{:noreply, %{state | initial_load_done: true}}
|
||||
end
|
||||
|
||||
def handle_info(:refresh_cache, state) do
|
||||
|
|
@ -101,33 +94,22 @@ defmodule Aprsme.DeviceCache do
|
|||
defp load_devices_into_cache do
|
||||
require Logger
|
||||
|
||||
# Skip database loading in test environment
|
||||
if Application.get_env(:aprsme, :env) == :test do
|
||||
Cache.put(@cache_name, :all_devices, [])
|
||||
:ok
|
||||
else
|
||||
try do
|
||||
devices =
|
||||
try do
|
||||
Repo.all(Devices)
|
||||
rescue
|
||||
error ->
|
||||
Logger.error("Failed to load devices from database: #{inspect(error)}")
|
||||
[]
|
||||
end
|
||||
|
||||
# Store all devices in cache
|
||||
case Cache.put(@cache_name, :all_devices, devices) do
|
||||
{:ok, true} -> :ok
|
||||
error -> error
|
||||
devices =
|
||||
try do
|
||||
Repo.all(Devices)
|
||||
rescue
|
||||
error ->
|
||||
Logger.error("Failed to load devices from database: #{inspect(error)}")
|
||||
[]
|
||||
end
|
||||
rescue
|
||||
error in [Postgrex.Error, DBConnection.ConnectionError] ->
|
||||
# Handle case where database or table doesn't exist yet
|
||||
Logger.warning("Failed to load devices: #{inspect(error)}. Will retry later.")
|
||||
# Store empty list for now
|
||||
Cache.put(@cache_name, :all_devices, [])
|
||||
:error
|
||||
|
||||
case Cache.put(@cache_name, :all_devices, devices) do
|
||||
{:ok, true} -> :ok
|
||||
error -> error
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -188,11 +188,6 @@ defmodule Aprsme.DeviceIdentification do
|
|||
|> Map.put("updated_at", now)
|
||||
end
|
||||
|
||||
# Helper to enqueue the job
|
||||
def enqueue_refresh_job do
|
||||
# Oban.insert!(Worker.new(%{}))
|
||||
end
|
||||
|
||||
@doc """
|
||||
Looks up a device by identifier, using ? as a single-character wildcard.
|
||||
Returns the device struct if found, or nil.
|
||||
|
|
|
|||
|
|
@ -137,21 +137,43 @@ defmodule Aprsme.Encoding do
|
|||
<<byte>>
|
||||
end
|
||||
|
||||
defp latin1_char_to_utf8(byte) when byte >= 128 and byte <= 159 do
|
||||
# C1 control range (128-159) has no useful Latin1 characters.
|
||||
# Replace with Unicode replacement character instead of converting
|
||||
# to C1 control codepoints that would just get stripped later.
|
||||
<<0xEF, 0xBF, 0xBD>>
|
||||
end
|
||||
|
||||
defp latin1_char_to_utf8(byte) do
|
||||
# For Latin1, values 128-255 map to Unicode U+0080 to U+00FF
|
||||
# For Latin1, values 160-255 map to Unicode U+00A0 to U+00FF
|
||||
# In UTF-8, these become 2-byte sequences: 110xxxxx 10xxxxxx
|
||||
<<0xC0 + div(byte, 64), 0x80 + rem(byte, 64)>>
|
||||
end
|
||||
|
||||
@spec clean_control_characters(binary()) :: binary()
|
||||
defp clean_control_characters(s) do
|
||||
s
|
||||
|> String.graphemes()
|
||||
|> Enum.filter(&valid_grapheme?/1)
|
||||
|> Enum.join()
|
||||
|> String.trim()
|
||||
if all_bytes_clean?(s) do
|
||||
String.trim(s)
|
||||
else
|
||||
s
|
||||
|> String.graphemes()
|
||||
|> Enum.filter(&valid_grapheme?/1)
|
||||
|> Enum.join()
|
||||
|> String.trim()
|
||||
end
|
||||
end
|
||||
|
||||
# Fast-path check: returns true when every byte is safe ASCII
|
||||
# (printable chars 32-126, plus tab/newline/CR). Any byte >= 128
|
||||
# or any control character triggers the slow grapheme-based path.
|
||||
@spec all_bytes_clean?(binary()) :: boolean()
|
||||
defp all_bytes_clean?(<<>>), do: true
|
||||
defp all_bytes_clean?(<<9, rest::binary>>), do: all_bytes_clean?(rest)
|
||||
defp all_bytes_clean?(<<10, rest::binary>>), do: all_bytes_clean?(rest)
|
||||
defp all_bytes_clean?(<<13, rest::binary>>), do: all_bytes_clean?(rest)
|
||||
defp all_bytes_clean?(<<byte, rest::binary>>) when byte >= 32 and byte <= 126, do: all_bytes_clean?(rest)
|
||||
defp all_bytes_clean?(_), do: false
|
||||
|
||||
@spec valid_grapheme?(String.grapheme()) :: boolean()
|
||||
defp valid_grapheme?(grapheme) do
|
||||
case String.to_charlist(grapheme) do
|
||||
|
|
|
|||
|
|
@ -1,53 +0,0 @@
|
|||
defmodule Aprsme.PacketPipelineSetup do
|
||||
@moduledoc """
|
||||
Handles the setup of the GenStage pipeline subscription after the supervisor starts.
|
||||
"""
|
||||
use GenServer
|
||||
|
||||
def start_link(opts \\ []) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def init(_opts) do
|
||||
# Wait a moment for the pipeline to start, then set up the subscription
|
||||
Process.send_after(self(), :setup_subscription, 100)
|
||||
{:ok, %{}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:setup_subscription, state) do
|
||||
# Set up the subscription between producer and consumer
|
||||
config = Application.get_env(:aprsme, :packet_pipeline, [])
|
||||
max_demand = config[:max_demand] || 50
|
||||
|
||||
try do
|
||||
case GenStage.sync_subscribe(Aprsme.PacketConsumer,
|
||||
to: Aprsme.PacketProducer,
|
||||
max_demand: max_demand
|
||||
) do
|
||||
{:ok, _subscription} ->
|
||||
require Logger
|
||||
|
||||
Logger.info("GenStage packet pipeline subscription established")
|
||||
{:noreply, state}
|
||||
|
||||
{:error, reason} ->
|
||||
require Logger
|
||||
|
||||
Logger.error("Failed to establish GenStage subscription: #{inspect(reason)}")
|
||||
# Retry after a delay
|
||||
Process.send_after(self(), :setup_subscription, 1000)
|
||||
{:noreply, state}
|
||||
end
|
||||
catch
|
||||
:exit, reason ->
|
||||
require Logger
|
||||
|
||||
Logger.error("Failed to establish GenStage subscription: #{inspect(reason)}")
|
||||
# Retry after a delay
|
||||
Process.send_after(self(), :setup_subscription, 1000)
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -72,25 +72,11 @@ defmodule Aprsme.Packets.QueryBuilder do
|
|||
|
||||
@doc """
|
||||
Filters query to only weather packets.
|
||||
Uses SQL to check if any weather field is not null.
|
||||
Uses the indexed `has_weather` boolean column.
|
||||
"""
|
||||
@spec weather_only(Ecto.Query.t()) :: Ecto.Query.t()
|
||||
def weather_only(query) do
|
||||
from p in query,
|
||||
where:
|
||||
fragment(
|
||||
"? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL OR ? IS NOT NULL",
|
||||
p.temperature,
|
||||
p.humidity,
|
||||
p.pressure,
|
||||
p.wind_speed,
|
||||
p.wind_direction,
|
||||
p.rain_1h,
|
||||
p.rain_24h,
|
||||
p.rain_since_midnight,
|
||||
p.snow,
|
||||
p.luminosity
|
||||
)
|
||||
from p in query, where: p.has_weather == true
|
||||
end
|
||||
|
||||
@doc """
|
||||
|
|
|
|||
|
|
@ -9,22 +9,4 @@ defmodule Aprsme.RateLimiterWrapper do
|
|||
def hit(bucket, scale_ms, limit) do
|
||||
Aprsme.RateLimiter.hit(bucket, scale_ms, limit)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Reset rate limit for a bucket
|
||||
"""
|
||||
def reset(_bucket) do
|
||||
# ETS-based rate limiter doesn't provide a reset function
|
||||
# Return ok
|
||||
:ok
|
||||
end
|
||||
|
||||
@doc """
|
||||
Get current count for a bucket
|
||||
"""
|
||||
def count(_bucket, _scale_ms) do
|
||||
# ETS-based rate limiter doesn't provide a count function
|
||||
# Return 0 as default
|
||||
{:ok, 0}
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -288,14 +288,22 @@ defmodule Aprsme.SpatialPubSub do
|
|||
end
|
||||
|
||||
defp get_intersecting_grid_cells(%{north: n, south: s, east: e, west: w}) do
|
||||
# Calculate which grid cells intersect with the bounds
|
||||
min_lat_cell = floor(s / @grid_size)
|
||||
max_lat_cell = floor(n / @grid_size)
|
||||
min_lon_cell = floor(w / @grid_size)
|
||||
max_lon_cell = floor(e / @grid_size)
|
||||
|
||||
lon_cells =
|
||||
if min_lon_cell > max_lon_cell do
|
||||
max_positive = floor(180.0 / @grid_size)
|
||||
min_negative = floor(-180.0 / @grid_size)
|
||||
Enum.to_list(min_lon_cell..max_positive) ++ Enum.to_list(min_negative..max_lon_cell)
|
||||
else
|
||||
Enum.to_list(min_lon_cell..max_lon_cell)
|
||||
end
|
||||
|
||||
for lat_cell <- min_lat_cell..max_lat_cell,
|
||||
lon_cell <- min_lon_cell..max_lon_cell do
|
||||
lon_cell <- lon_cells do
|
||||
{lat_cell, lon_cell}
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,298 +0,0 @@
|
|||
defmodule Aprsme.SystemMonitor do
|
||||
@moduledoc """
|
||||
Monitors system metrics to help with adaptive performance tuning.
|
||||
"""
|
||||
use GenServer
|
||||
|
||||
require Logger
|
||||
|
||||
@check_interval 5_000
|
||||
@min_batch_size 100
|
||||
@max_batch_size 800
|
||||
@default_batch_size 200
|
||||
|
||||
def start_link(opts \\ []) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
def get_recommended_batch_size do
|
||||
GenServer.call(__MODULE__, :get_batch_size)
|
||||
catch
|
||||
:exit, {:noproc, _} -> @default_batch_size
|
||||
end
|
||||
|
||||
def get_metrics do
|
||||
GenServer.call(__MODULE__, :get_metrics)
|
||||
catch
|
||||
:exit, {:noproc, _} -> default_metrics()
|
||||
end
|
||||
|
||||
@impl true
|
||||
def init(_opts) do
|
||||
schedule_check()
|
||||
|
||||
state = %{
|
||||
metrics: default_metrics(),
|
||||
batch_size: @default_batch_size,
|
||||
history: []
|
||||
}
|
||||
|
||||
{:ok, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:get_batch_size, _from, state) do
|
||||
{:reply, state.batch_size, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:get_metrics, _from, state) do
|
||||
{:reply, state.metrics, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:check_system, state) do
|
||||
metrics = collect_metrics()
|
||||
new_batch_size = calculate_optimal_batch_size(metrics, state)
|
||||
|
||||
# Keep history for trend analysis (last 12 data points = 1 minute)
|
||||
history = Enum.take([metrics | state.history], 12)
|
||||
|
||||
# Emit telemetry events for LiveDashboard
|
||||
emit_telemetry_events(metrics, new_batch_size)
|
||||
|
||||
new_state = %{state | metrics: metrics, batch_size: new_batch_size, history: history}
|
||||
|
||||
schedule_check()
|
||||
{:noreply, new_state}
|
||||
end
|
||||
|
||||
defp schedule_check do
|
||||
Process.send_after(self(), :check_system, @check_interval)
|
||||
end
|
||||
|
||||
defp collect_metrics do
|
||||
# Memory metrics
|
||||
memory_info = :erlang.memory()
|
||||
total_memory = memory_info[:total]
|
||||
process_memory = memory_info[:processes]
|
||||
binary_memory = memory_info[:binary]
|
||||
|
||||
# CPU metrics
|
||||
scheduler_count = :erlang.system_info(:schedulers_online)
|
||||
|
||||
# Parse load averages more robustly
|
||||
load_values =
|
||||
~c"uptime | awk -F'load average:' '{print $2}'"
|
||||
|> :os.cmd()
|
||||
|> to_string()
|
||||
|> String.trim()
|
||||
|> String.split(",")
|
||||
|> Enum.map(&String.trim/1)
|
||||
|> Enum.map(&parse_float/1)
|
||||
|
||||
# Ensure we have 3 values, defaulting to 0.0 if missing
|
||||
{load1, load5, load15} =
|
||||
case load_values do
|
||||
[l1, l5, l15 | _] -> {l1, l5, l15}
|
||||
[l1, l5] -> {l1, l5, 0.0}
|
||||
[l1] -> {l1, 0.0, 0.0}
|
||||
[] -> {0.0, 0.0, 0.0}
|
||||
end
|
||||
|
||||
# Process metrics
|
||||
process_count = :erlang.system_info(:process_count)
|
||||
|
||||
# Database pool metrics
|
||||
db_pool_status = get_db_pool_status()
|
||||
|
||||
# Calculate memory pressure (0.0 to 1.0)
|
||||
memory_pressure = calculate_memory_pressure(memory_info)
|
||||
|
||||
# Calculate CPU pressure (0.0 to 1.0)
|
||||
cpu_pressure = min(1.0, load1 / scheduler_count)
|
||||
|
||||
%{
|
||||
memory: %{
|
||||
total: total_memory,
|
||||
process: process_memory,
|
||||
binary: binary_memory,
|
||||
pressure: memory_pressure
|
||||
},
|
||||
cpu: %{
|
||||
load1: load1,
|
||||
load5: load5,
|
||||
load15: load15,
|
||||
schedulers: scheduler_count,
|
||||
pressure: cpu_pressure
|
||||
},
|
||||
processes: %{
|
||||
count: process_count,
|
||||
pressure: min(1.0, process_count / 50_000)
|
||||
},
|
||||
db_pool: db_pool_status,
|
||||
timestamp: DateTime.utc_now()
|
||||
}
|
||||
end
|
||||
|
||||
defp parse_float(str) do
|
||||
case Float.parse(str) do
|
||||
{float, _} -> float
|
||||
:error -> 0.0
|
||||
end
|
||||
end
|
||||
|
||||
defp calculate_memory_pressure(memory_info) do
|
||||
# Get app memory for pressure calculation
|
||||
app_memory = memory_info[:total]
|
||||
|
||||
# Assume 4GB available memory as baseline
|
||||
available_memory = 4 * 1024 * 1024 * 1024
|
||||
|
||||
# Calculate pressure based on usage
|
||||
min(1.0, app_memory / available_memory)
|
||||
end
|
||||
|
||||
defp get_db_pool_status do
|
||||
pool_config = Aprsme.Repo.config()[:pool_size] || 10
|
||||
|
||||
# Get pool telemetry if available
|
||||
:telemetry.execute([:aprsme, :repo, :pool], %{}, %{})
|
||||
|
||||
%{
|
||||
size: pool_config,
|
||||
# Would need actual telemetry
|
||||
available: pool_config,
|
||||
# Placeholder
|
||||
pressure: 0.3
|
||||
}
|
||||
rescue
|
||||
_ -> %{size: 10, available: 7, pressure: 0.3}
|
||||
end
|
||||
|
||||
defp calculate_optimal_batch_size(metrics, state) do
|
||||
# Base factors
|
||||
memory_factor = 1.0 - metrics.memory.pressure
|
||||
cpu_factor = 1.0 - metrics.cpu.pressure
|
||||
db_factor = 1.0 - metrics.db_pool.pressure
|
||||
|
||||
# Historical trend analysis
|
||||
trend_factor = calculate_trend_factor(state.history)
|
||||
|
||||
# Weighted combination
|
||||
combined_factor =
|
||||
memory_factor * 0.4 +
|
||||
cpu_factor * 0.3 +
|
||||
db_factor * 0.2 +
|
||||
trend_factor * 0.1
|
||||
|
||||
# Calculate new batch size
|
||||
target_size =
|
||||
@min_batch_size +
|
||||
round(combined_factor * (@max_batch_size - @min_batch_size))
|
||||
|
||||
# Apply smoothing to avoid rapid changes
|
||||
current_size = state.batch_size
|
||||
step = round((target_size - current_size) * 0.3)
|
||||
new_size = current_size + step
|
||||
|
||||
# Ensure within bounds
|
||||
new_size
|
||||
|> max(@min_batch_size)
|
||||
|> min(@max_batch_size)
|
||||
end
|
||||
|
||||
defp calculate_trend_factor(history) when length(history) < 3, do: 0.5
|
||||
|
||||
defp calculate_trend_factor(history) do
|
||||
# Analyze recent pressure trends
|
||||
recent_pressures =
|
||||
history
|
||||
|> Enum.take(3)
|
||||
|> Enum.map(fn m ->
|
||||
(m.memory.pressure + m.cpu.pressure + m.db_pool.pressure) / 3
|
||||
end)
|
||||
|
||||
case recent_pressures do
|
||||
[p1, p2, p3] when p1 > p2 and p2 > p3 ->
|
||||
# Pressure increasing, reduce batch size
|
||||
0.2
|
||||
|
||||
[p1, p2, p3] when p1 < p2 and p2 < p3 ->
|
||||
# Pressure decreasing, increase batch size
|
||||
0.8
|
||||
|
||||
_ ->
|
||||
# Stable
|
||||
0.5
|
||||
end
|
||||
end
|
||||
|
||||
defp default_metrics do
|
||||
%{
|
||||
memory: %{total: 0, process: 0, binary: 0, pressure: 0.5},
|
||||
cpu: %{load1: 1.0, load5: 1.0, load15: 1.0, schedulers: 1, pressure: 0.5},
|
||||
processes: %{count: 1000, pressure: 0.5},
|
||||
db_pool: %{size: 10, available: 7, pressure: 0.3},
|
||||
timestamp: DateTime.utc_now()
|
||||
}
|
||||
end
|
||||
|
||||
defp emit_telemetry_events(metrics, batch_size) do
|
||||
# Memory metrics
|
||||
:telemetry.execute(
|
||||
[:aprsme, :system, :memory],
|
||||
%{
|
||||
total: metrics.memory.total,
|
||||
process: metrics.memory.process,
|
||||
binary: metrics.memory.binary,
|
||||
pressure: metrics.memory.pressure
|
||||
},
|
||||
%{}
|
||||
)
|
||||
|
||||
# CPU metrics
|
||||
:telemetry.execute(
|
||||
[:aprsme, :system, :cpu],
|
||||
%{
|
||||
load1: metrics.cpu.load1,
|
||||
load5: metrics.cpu.load5,
|
||||
load15: metrics.cpu.load15,
|
||||
pressure: metrics.cpu.pressure
|
||||
},
|
||||
%{schedulers: metrics.cpu.schedulers}
|
||||
)
|
||||
|
||||
# Process metrics
|
||||
:telemetry.execute(
|
||||
[:aprsme, :system, :processes],
|
||||
%{
|
||||
count: metrics.processes.count,
|
||||
pressure: metrics.processes.pressure
|
||||
},
|
||||
%{}
|
||||
)
|
||||
|
||||
# Database pool metrics
|
||||
:telemetry.execute(
|
||||
[:aprsme, :system, :db_pool],
|
||||
%{
|
||||
size: metrics.db_pool.size,
|
||||
available: metrics.db_pool.available,
|
||||
pressure: metrics.db_pool.pressure
|
||||
},
|
||||
%{}
|
||||
)
|
||||
|
||||
# Batch size metrics
|
||||
:telemetry.execute(
|
||||
[:aprsme, :system, :batch_size],
|
||||
%{
|
||||
current: batch_size,
|
||||
min: @min_batch_size,
|
||||
max: @max_batch_size
|
||||
},
|
||||
%{}
|
||||
)
|
||||
end
|
||||
end
|
||||
|
|
@ -9,6 +9,7 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
|
|||
alias AprsmeWeb.Live.Shared.CoordinateUtils
|
||||
alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils
|
||||
alias AprsmeWeb.MapLive.DataBuilder
|
||||
alias AprsmeWeb.MapLive.DisplayManager
|
||||
alias AprsmeWeb.MapLive.RfPath
|
||||
alias Phoenix.LiveView
|
||||
alias Phoenix.LiveView.Socket
|
||||
|
|
@ -412,8 +413,7 @@ defmodule AprsmeWeb.MapLive.HistoricalLoader do
|
|||
end
|
||||
|
||||
defp send_heat_map_for_current_bounds(socket) do
|
||||
# This function should be moved to DisplayManager module
|
||||
socket
|
||||
DisplayManager.send_heat_map_for_current_bounds(socket)
|
||||
end
|
||||
|
||||
defp maybe_load_rf_path_stations(socket, batch_offset, historical_packets) do
|
||||
|
|
|
|||
|
|
@ -200,6 +200,7 @@ defmodule AprsmeWeb.MapLive.Index do
|
|||
|
||||
# Start packet batcher for efficient updates
|
||||
{:ok, batcher_pid} = PacketBatcher.start_link(self())
|
||||
Process.monitor(batcher_pid)
|
||||
|
||||
# Determine initial slideover state from client viewport width
|
||||
slideover_open = initial_slideover_open?(socket)
|
||||
|
|
@ -870,6 +871,13 @@ defmodule AprsmeWeb.MapLive.Index do
|
|||
{:noreply, socket}
|
||||
end
|
||||
|
||||
def handle_info({:DOWN, _ref, :process, pid, _reason}, socket) when pid == socket.assigns.batcher_pid do
|
||||
# PacketBatcher crashed — restart it
|
||||
{:ok, new_pid} = PacketBatcher.start_link(self())
|
||||
Process.monitor(new_pid)
|
||||
{:noreply, assign(socket, :batcher_pid, new_pid)}
|
||||
end
|
||||
|
||||
def handle_info(:clear_rf_path, socket) do
|
||||
# Clear the RF path lines with debouncing
|
||||
{:noreply, push_event(socket, "clear_rf_path", %{})}
|
||||
|
|
@ -1773,9 +1781,7 @@ defmodule AprsmeWeb.MapLive.Index do
|
|||
# Schedule next update
|
||||
Process.send_after(self(), :update_time_display, 30_000)
|
||||
|
||||
# Simply triggering a re-render will cause time_ago_in_words to recalculate
|
||||
# No need to update any assigns, just return the socket
|
||||
{:noreply, socket}
|
||||
{:noreply, assign(socket, :time_display_tick, System.monotonic_time())}
|
||||
end
|
||||
|
||||
defp handle_reload_historical_packets(socket) do
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
|||
alias AprsmeWeb.Live.Shared.BoundsUtils
|
||||
alias AprsmeWeb.Live.Shared.CoordinateUtils
|
||||
alias AprsmeWeb.Live.Shared.PacketUtils, as: SharedPacketUtils
|
||||
alias AprsmeWeb.MapLive.DisplayManager
|
||||
alias AprsmeWeb.MapLive.PacketUtils
|
||||
alias Phoenix.LiveView
|
||||
alias Phoenix.LiveView.Socket
|
||||
|
|
@ -143,10 +144,8 @@ defmodule AprsmeWeb.MapLive.PacketProcessor do
|
|||
Map.get(socket.assigns, :locale, "en")
|
||||
end
|
||||
|
||||
# Placeholder for heat map function - this should be moved to DisplayManager
|
||||
defp send_heat_map_for_current_bounds(socket) do
|
||||
# This function should be moved to DisplayManager module
|
||||
socket
|
||||
DisplayManager.send_heat_map_for_current_bounds(socket)
|
||||
end
|
||||
|
||||
defp send_marker_with_popup_check(socket, marker_data) do
|
||||
|
|
|
|||
|
|
@ -1,238 +0,0 @@
|
|||
defmodule Aprsme.ArchiverTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias Aprsme.Archiver
|
||||
|
||||
describe "start_link/1" do
|
||||
test "starts the archiver GenServer with default args" do
|
||||
assert {:ok, pid} = Archiver.start_link()
|
||||
assert is_pid(pid)
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "starts the archiver GenServer with custom args" do
|
||||
args = [some: :option]
|
||||
assert {:ok, pid} = Archiver.start_link(args)
|
||||
assert is_pid(pid)
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "registers the process with name :archiver" do
|
||||
# Ensure no previous archiver is running
|
||||
if Process.whereis(:archiver) do
|
||||
GenServer.stop(:archiver)
|
||||
Process.sleep(10)
|
||||
end
|
||||
|
||||
assert {:ok, pid} = Archiver.start_link()
|
||||
assert Process.whereis(:archiver) == pid
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(:archiver)
|
||||
end
|
||||
|
||||
test "returns error if archiver is already running" do
|
||||
# Ensure no previous archiver is running
|
||||
if Process.whereis(:archiver) do
|
||||
GenServer.stop(:archiver)
|
||||
Process.sleep(10)
|
||||
end
|
||||
|
||||
assert {:ok, _pid1} = Archiver.start_link()
|
||||
assert {:error, {:already_started, _pid}} = Archiver.start_link()
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(:archiver)
|
||||
end
|
||||
end
|
||||
|
||||
describe "init/1" do
|
||||
test "initializes with default empty state" do
|
||||
assert {:ok, []} = Archiver.init()
|
||||
end
|
||||
|
||||
test "initializes with provided state" do
|
||||
state = [some: :data]
|
||||
assert {:ok, ^state} = Archiver.init(state)
|
||||
end
|
||||
|
||||
test "schedules connect message after 5 seconds" do
|
||||
# Start the GenServer to test init behavior
|
||||
{:ok, pid} = Archiver.start_link()
|
||||
|
||||
# We can't easily test the exact timing without making the test slow,
|
||||
# but we can verify the GenServer started successfully
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "subscribes to the 'call' topic" do
|
||||
# Mock the Endpoint.subscribe call
|
||||
# Since we can't easily mock in this context, we'll just verify
|
||||
# the GenServer starts without errors
|
||||
{:ok, pid} = Archiver.start_link()
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
end
|
||||
|
||||
describe "handle_info/2" do
|
||||
setup do
|
||||
{:ok, pid} = Archiver.start_link()
|
||||
%{pid: pid}
|
||||
end
|
||||
|
||||
test "handles :connect message", %{pid: pid} do
|
||||
state = [some: :state]
|
||||
|
||||
# Send the connect message directly to test the handler
|
||||
result = Archiver.handle_info(:connect, state)
|
||||
|
||||
assert {:noreply, ^state} = result
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "handles unknown messages", %{pid: pid} do
|
||||
state = [some: :state]
|
||||
|
||||
# Send an unknown message
|
||||
result = Archiver.handle_info(:unknown_message, state)
|
||||
|
||||
assert {:noreply, ^state} = result
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "handles various message types", %{pid: pid} do
|
||||
state = [test: :data]
|
||||
|
||||
# Test different message types
|
||||
messages = [
|
||||
{:some, :tuple},
|
||||
"string_message",
|
||||
123,
|
||||
%{map: "message"},
|
||||
[:list, :message]
|
||||
]
|
||||
|
||||
for message <- messages do
|
||||
result = Archiver.handle_info(message, state)
|
||||
assert {:noreply, ^state} = result
|
||||
end
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "preserves state across different messages", %{pid: pid} do
|
||||
initial_state = [counter: 0]
|
||||
|
||||
# Test that state is preserved
|
||||
result1 = Archiver.handle_info(:connect, initial_state)
|
||||
assert {:noreply, ^initial_state} = result1
|
||||
|
||||
result2 = Archiver.handle_info(:other_message, initial_state)
|
||||
assert {:noreply, ^initial_state} = result2
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
end
|
||||
|
||||
describe "GenServer behavior" do
|
||||
test "can send messages to running archiver" do
|
||||
{:ok, pid} = Archiver.start_link()
|
||||
|
||||
# Send a message and verify the process handles it
|
||||
send(pid, :test_message)
|
||||
|
||||
# Give it a moment to process
|
||||
Process.sleep(10)
|
||||
|
||||
# Verify the process is still alive (didn't crash)
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "can be stopped gracefully" do
|
||||
{:ok, pid} = Archiver.start_link()
|
||||
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Stop the GenServer
|
||||
:ok = GenServer.stop(pid)
|
||||
|
||||
# Give it a moment to stop
|
||||
Process.sleep(10)
|
||||
|
||||
# Verify it's no longer alive
|
||||
refute Process.alive?(pid)
|
||||
end
|
||||
|
||||
test "handles multiple concurrent operations" do
|
||||
{:ok, pid} = Archiver.start_link()
|
||||
|
||||
# Send multiple messages concurrently
|
||||
tasks =
|
||||
for i <- 1..10 do
|
||||
Task.async(fn ->
|
||||
send(pid, {:message, i})
|
||||
:ok
|
||||
end)
|
||||
end
|
||||
|
||||
# Wait for all tasks to complete
|
||||
Enum.each(tasks, &Task.await/1)
|
||||
|
||||
# Verify the process is still alive
|
||||
assert Process.alive?(pid)
|
||||
|
||||
# Clean up
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
end
|
||||
|
||||
describe "module constants and specs" do
|
||||
test "has correct topic constant" do
|
||||
# This tests that the module compiles correctly with the @topic attribute
|
||||
# The actual value is tested indirectly through the init function
|
||||
assert Code.ensure_loaded?(Archiver)
|
||||
end
|
||||
|
||||
test "has correct typespec for start_link" do
|
||||
# This ensures the module compiles with correct typespecs
|
||||
# We can't directly test typespecs, but we can ensure the function
|
||||
# behaves according to its spec
|
||||
result = Archiver.start_link([])
|
||||
assert match?({:ok, _pid}, result) or match?({:error, _reason}, result)
|
||||
|
||||
if match?({:ok, _pid}, result) do
|
||||
GenServer.stop(elem(result, 1))
|
||||
end
|
||||
end
|
||||
|
||||
test "has correct typespec for init" do
|
||||
# Test that init returns the expected format
|
||||
result = Archiver.init([])
|
||||
assert {:ok, []} = result
|
||||
|
||||
result2 = Archiver.init(some: :data)
|
||||
assert {:ok, [some: :data]} = result2
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -45,10 +45,42 @@ defmodule Aprsme.EncodingTest do
|
|||
end
|
||||
|
||||
test "handles latin1 encoded binary" do
|
||||
# Latin1 bytes for accented characters
|
||||
# Latin1 bytes for accented characters (>= 160, valid Latin1 printable)
|
||||
invalid_binary = <<85, 78, 73, 211, 78, 32, 80, 65, 78, 65, 77, 69, 209, 65>>
|
||||
result = Encoding.sanitize_string(invalid_binary)
|
||||
assert String.valid?(result)
|
||||
# 211 = Ó, 209 = Ñ — both above 159, so they convert to real characters
|
||||
assert result == "UNIÓN PANAMEÑA"
|
||||
end
|
||||
end
|
||||
|
||||
describe "sanitize_string/1 C1 control range (128-159)" do
|
||||
test "latin1 bytes 128-159 become replacement characters instead of being silently deleted" do
|
||||
input = <<72, 101, 108, 108, 111, 130>>
|
||||
result = Encoding.sanitize_string(input)
|
||||
assert String.valid?(result)
|
||||
assert String.contains?(result, "\uFFFD")
|
||||
end
|
||||
|
||||
test "latin1 byte 128 becomes replacement character" do
|
||||
input = <<65, 128, 66>>
|
||||
result = Encoding.sanitize_string(input)
|
||||
assert String.valid?(result)
|
||||
assert result == "A\uFFFDB"
|
||||
end
|
||||
|
||||
test "latin1 byte 159 becomes replacement character" do
|
||||
input = <<65, 159, 66>>
|
||||
result = Encoding.sanitize_string(input)
|
||||
assert String.valid?(result)
|
||||
assert result == "A\uFFFDB"
|
||||
end
|
||||
|
||||
test "latin1 bytes above 159 still convert to proper characters" do
|
||||
input = <<67, 97, 102, 233>>
|
||||
result = Encoding.sanitize_string(input)
|
||||
assert String.valid?(result)
|
||||
assert result == "Café"
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
|
|
@ -1,99 +0,0 @@
|
|||
defmodule Aprsme.PacketPipelineSetupTest do
|
||||
use ExUnit.Case, async: false
|
||||
|
||||
import ExUnit.CaptureLog
|
||||
|
||||
alias Aprsme.PacketPipelineSetup
|
||||
|
||||
setup do
|
||||
# Stop any existing PacketPipelineSetup process
|
||||
case GenServer.whereis(PacketPipelineSetup) do
|
||||
nil -> :ok
|
||||
pid -> GenServer.stop(pid, :normal, 5000)
|
||||
end
|
||||
|
||||
on_exit(fn ->
|
||||
case GenServer.whereis(PacketPipelineSetup) do
|
||||
nil ->
|
||||
:ok
|
||||
|
||||
pid ->
|
||||
try do
|
||||
GenServer.stop(pid, :normal, 5000)
|
||||
catch
|
||||
:exit, _ -> :ok
|
||||
end
|
||||
end
|
||||
end)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
describe "start_link/0" do
|
||||
test "starts the GenServer with default opts" do
|
||||
capture_log(fn ->
|
||||
{:ok, pid} = PacketPipelineSetup.start_link()
|
||||
|
||||
assert Process.alive?(pid)
|
||||
assert GenServer.whereis(PacketPipelineSetup) == pid
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
describe "start_link/1" do
|
||||
test "starts the GenServer and registers it under its module name" do
|
||||
capture_log(fn ->
|
||||
{:ok, pid} = PacketPipelineSetup.start_link([])
|
||||
|
||||
assert Process.alive?(pid)
|
||||
assert GenServer.whereis(PacketPipelineSetup) == pid
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
describe "handle_info(:setup_subscription, state)" do
|
||||
test "GenServer survives when PacketProducer/PacketConsumer are not running and schedules retry" do
|
||||
log =
|
||||
capture_log(fn ->
|
||||
{:ok, pid} = PacketPipelineSetup.start_link([])
|
||||
|
||||
# Wait for the initial :setup_subscription to fire (~100ms) and fail
|
||||
Process.sleep(200)
|
||||
|
||||
# The GenServer should still be alive after the failure
|
||||
assert Process.alive?(pid)
|
||||
end)
|
||||
|
||||
# The error path should have logged the failure
|
||||
assert log =~ "Failed to establish GenStage subscription"
|
||||
end
|
||||
|
||||
test "retries subscription after failure" do
|
||||
log =
|
||||
capture_log(fn ->
|
||||
{:ok, pid} = PacketPipelineSetup.start_link([])
|
||||
|
||||
# Wait for initial attempt (~100ms) and retry (~1000ms)
|
||||
Process.sleep(1300)
|
||||
|
||||
assert Process.alive?(pid)
|
||||
end)
|
||||
|
||||
# Should see multiple failure logs from retry
|
||||
assert log =~ "Failed to establish GenStage subscription"
|
||||
end
|
||||
end
|
||||
|
||||
describe "init/1" do
|
||||
test "schedules :setup_subscription message" do
|
||||
capture_log(fn ->
|
||||
{:ok, pid} = PacketPipelineSetup.start_link([])
|
||||
|
||||
# The init should have scheduled a message
|
||||
# Verify state is initialized as empty map
|
||||
state = :sys.get_state(pid)
|
||||
assert state == %{}
|
||||
end)
|
||||
end
|
||||
end
|
||||
end
|
||||
61
test/aprsme/spatial_pubsub_test.exs
Normal file
61
test/aprsme/spatial_pubsub_test.exs
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
defmodule Aprsme.SpatialPubSubTest do
|
||||
use ExUnit.Case, async: false
|
||||
|
||||
alias Aprsme.SpatialPubSub
|
||||
|
||||
describe "viewport registration" do
|
||||
test "date-line-crossing viewport registers without timeout" do
|
||||
bounds = %{north: 10.0, south: -10.0, east: -170.0, west: 170.0}
|
||||
client_id = "test_dateline_#{:rand.uniform(100_000)}"
|
||||
|
||||
# Before the fix, west=170 east=-170 generated ~341 grid cells.
|
||||
# Now it correctly generates ~21 cells.
|
||||
assert {:ok, _topic} = SpatialPubSub.register_viewport(client_id, bounds)
|
||||
|
||||
SpatialPubSub.unregister_client(client_id)
|
||||
end
|
||||
|
||||
test "normal viewport registers successfully" do
|
||||
bounds = %{north: 41.0, south: 40.0, east: -73.0, west: -74.0}
|
||||
client_id = "test_normal_#{:rand.uniform(100_000)}"
|
||||
|
||||
assert {:ok, _topic} = SpatialPubSub.register_viewport(client_id, bounds)
|
||||
|
||||
SpatialPubSub.unregister_client(client_id)
|
||||
end
|
||||
|
||||
test "update_viewport succeeds after registration" do
|
||||
bounds = %{north: 41.0, south: 40.0, east: -73.0, west: -74.0}
|
||||
new_bounds = %{north: 42.0, south: 41.0, east: -72.0, west: -73.0}
|
||||
client_id = "test_update_#{:rand.uniform(100_000)}"
|
||||
|
||||
{:ok, _topic} = SpatialPubSub.register_viewport(client_id, bounds)
|
||||
assert :ok = SpatialPubSub.update_viewport(client_id, new_bounds)
|
||||
|
||||
SpatialPubSub.unregister_client(client_id)
|
||||
end
|
||||
end
|
||||
|
||||
describe "stats" do
|
||||
test "get_stats returns expected keys" do
|
||||
stats = SpatialPubSub.get_stats()
|
||||
|
||||
assert is_map(stats)
|
||||
assert Map.has_key?(stats, :total_packets)
|
||||
assert Map.has_key?(stats, :grid_cells)
|
||||
end
|
||||
|
||||
test "grid cells increase after registering viewport" do
|
||||
bounds = %{north: 41.0, south: 40.0, east: -73.0, west: -74.0}
|
||||
client_id = "test_stats_#{:rand.uniform(100_000)}"
|
||||
|
||||
before_stats = SpatialPubSub.get_stats()
|
||||
{:ok, _topic} = SpatialPubSub.register_viewport(client_id, bounds)
|
||||
after_stats = SpatialPubSub.get_stats()
|
||||
|
||||
assert after_stats.grid_cells >= before_stats.grid_cells
|
||||
|
||||
SpatialPubSub.unregister_client(client_id)
|
||||
end
|
||||
end
|
||||
end
|
||||
85
test/aprsme_web/live/map_live/packet_batcher_test.exs
Normal file
85
test/aprsme_web/live/map_live/packet_batcher_test.exs
Normal file
|
|
@ -0,0 +1,85 @@
|
|||
defmodule AprsmeWeb.MapLive.PacketBatcherTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias AprsmeWeb.MapLive.PacketBatcher
|
||||
|
||||
setup do
|
||||
# Trap exits so linked batcher doesn't crash the test process
|
||||
Process.flag(:trap_exit, true)
|
||||
:ok
|
||||
end
|
||||
|
||||
describe "start_link/1" do
|
||||
test "starts a batcher process" do
|
||||
{:ok, pid} = PacketBatcher.start_link(self())
|
||||
assert Process.alive?(pid)
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
end
|
||||
|
||||
describe "batching behavior" do
|
||||
test "delivers packets in batch after timeout" do
|
||||
{:ok, pid} = PacketBatcher.start_link(self())
|
||||
|
||||
PacketBatcher.add_packet(pid, %{sender: "TEST-1"})
|
||||
PacketBatcher.add_packet(pid, %{sender: "TEST-2"})
|
||||
|
||||
# Should receive batch after the 100ms timeout
|
||||
assert_receive {:packet_batch, packets}, 500
|
||||
assert length(packets) == 2
|
||||
assert Enum.at(packets, 0).sender == "TEST-1"
|
||||
assert Enum.at(packets, 1).sender == "TEST-2"
|
||||
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "delivers immediately when batch size reached" do
|
||||
{:ok, pid} = PacketBatcher.start_link(self())
|
||||
|
||||
# Send 10 packets (batch size threshold)
|
||||
for i <- 1..10 do
|
||||
PacketBatcher.add_packet(pid, %{sender: "TEST-#{i}"})
|
||||
end
|
||||
|
||||
# Should receive immediately (not waiting for timeout)
|
||||
assert_receive {:packet_batch, packets}, 200
|
||||
assert length(packets) == 10
|
||||
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
|
||||
test "flush/1 delivers buffered packets immediately" do
|
||||
{:ok, pid} = PacketBatcher.start_link(self())
|
||||
|
||||
PacketBatcher.add_packet(pid, %{sender: "TEST-1"})
|
||||
PacketBatcher.flush(pid)
|
||||
|
||||
assert_receive {:packet_batch, [%{sender: "TEST-1"}]}, 200
|
||||
|
||||
GenServer.stop(pid)
|
||||
end
|
||||
end
|
||||
|
||||
describe "crash recovery" do
|
||||
test "batcher stops when parent dies" do
|
||||
# Start a temporary parent process that traps exits
|
||||
parent =
|
||||
spawn(fn ->
|
||||
Process.flag(:trap_exit, true)
|
||||
|
||||
receive do
|
||||
:stop -> :ok
|
||||
end
|
||||
end)
|
||||
|
||||
{:ok, batcher_pid} = PacketBatcher.start_link(parent)
|
||||
ref = Process.monitor(batcher_pid)
|
||||
|
||||
# Kill the parent
|
||||
Process.exit(parent, :kill)
|
||||
|
||||
# Batcher should stop
|
||||
assert_receive {:DOWN, ^ref, :process, ^batcher_pid, :normal}, 1000
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue