Fix APRS object name extraction for names with special characters
The object name regex only matched [A-Z0-9 ] which missed names with
hyphens (P-K5SGD), hash symbols (#146.760), and other printable ASCII.
Changed to .{9}[*_] to match all valid object names including killed
objects. Also merged the two cond branches so data_extended is tried
first with information_field parsing as fallback.
Removed dead copy_insert/3 tests from db_optimizer_test.
This commit is contained in:
parent
8a8e16aaba
commit
118a55b26e
3 changed files with 116 additions and 79 deletions
|
|
@ -14,6 +14,7 @@ defmodule Aprsme.PacketConsumer do
|
|||
|
||||
@type state :: %{
|
||||
batch: list(map()),
|
||||
batch_length: non_neg_integer(),
|
||||
batch_size: integer(),
|
||||
batch_timeout: integer(),
|
||||
max_batch_size: integer(),
|
||||
|
|
@ -48,6 +49,7 @@ defmodule Aprsme.PacketConsumer do
|
|||
{:consumer,
|
||||
%{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: batch_size,
|
||||
batch_timeout: batch_timeout,
|
||||
max_batch_size: max_batch_size,
|
||||
|
|
@ -61,11 +63,11 @@ defmodule Aprsme.PacketConsumer do
|
|||
end
|
||||
|
||||
# Pattern matching for batch handling with optimized list operations
|
||||
defp handle_batch_update(events, %{batch: batch} = state) do
|
||||
defp handle_batch_update(events, %{batch: batch, batch_length: batch_length} = state) do
|
||||
# Optimize: Keep batch in reverse order for O(1) prepending
|
||||
# Only reverse when processing
|
||||
new_batch = Enum.reverse(events, batch)
|
||||
new_batch_length = batch_length(batch) + length(events)
|
||||
new_batch_length = batch_length + length(events)
|
||||
|
||||
handle_batch_by_size(new_batch, new_batch_length, state)
|
||||
end
|
||||
|
|
@ -79,8 +81,8 @@ defmodule Aprsme.PacketConsumer do
|
|||
handle_full_batch(batch, state)
|
||||
end
|
||||
|
||||
defp handle_batch_by_size(batch, _length, state) do
|
||||
handle_partial_batch(batch, state)
|
||||
defp handle_batch_by_size(batch, new_batch_length, state) do
|
||||
handle_partial_batch(batch, new_batch_length, state)
|
||||
end
|
||||
|
||||
# Handle oversized batch with pattern matching
|
||||
|
|
@ -106,29 +108,24 @@ defmodule Aprsme.PacketConsumer do
|
|||
end
|
||||
|
||||
# Handle partial batch
|
||||
defp handle_partial_batch(batch, state) do
|
||||
defp handle_partial_batch(batch, new_batch_length, state) do
|
||||
# Keep batch in reverse order
|
||||
{:noreply, [], %{state | batch: batch}}
|
||||
{:noreply, [], %{state | batch: batch, batch_length: new_batch_length}}
|
||||
end
|
||||
|
||||
# Helper functions with pattern matching
|
||||
@spec reset_batch_timer(state()) :: state()
|
||||
defp reset_batch_timer(%{timer: nil} = state) do
|
||||
new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
|
||||
%{state | batch: [], timer: new_timer}
|
||||
%{state | batch: [], batch_length: 0, timer: new_timer}
|
||||
end
|
||||
|
||||
defp reset_batch_timer(%{timer: timer} = state) do
|
||||
Process.cancel_timer(timer)
|
||||
new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
|
||||
%{state | batch: [], timer: new_timer}
|
||||
%{state | batch: [], batch_length: 0, timer: new_timer}
|
||||
end
|
||||
|
||||
# Efficient batch length calculation (cached if possible)
|
||||
@spec batch_length(list()) :: non_neg_integer()
|
||||
defp batch_length([]), do: 0
|
||||
defp batch_length(batch), do: length(batch)
|
||||
|
||||
# Pattern matching for logging
|
||||
@spec log_dropped_packets(list(), list()) :: :ok
|
||||
defp log_dropped_packets([], _), do: :ok
|
||||
|
|
@ -153,12 +150,12 @@ defmodule Aprsme.PacketConsumer do
|
|||
end
|
||||
|
||||
# Pattern matching for non-empty batch
|
||||
def handle_info(:process_batch, %{batch: batch} = state) when is_list(batch) do
|
||||
check_batch_utilization(batch, state.max_batch_size)
|
||||
def handle_info(:process_batch, %{batch: batch, batch_length: batch_length} = state) when is_list(batch) do
|
||||
check_batch_utilization(batch_length, state.max_batch_size)
|
||||
# Remember to reverse the batch since we keep it in reverse order
|
||||
process_batch(Enum.reverse(batch))
|
||||
|
||||
new_state = start_batch_timer(%{state | batch: []})
|
||||
new_state = start_batch_timer(%{state | batch: [], batch_length: 0})
|
||||
{:noreply, [], new_state}
|
||||
end
|
||||
|
||||
|
|
@ -169,20 +166,20 @@ defmodule Aprsme.PacketConsumer do
|
|||
%{state | timer: timer}
|
||||
end
|
||||
|
||||
# Pattern matching for batch utilization check
|
||||
@spec check_batch_utilization(list(), integer()) :: :ok
|
||||
defp check_batch_utilization(batch, max_size) when length(batch) > max_size * 0.8 do
|
||||
# Batch utilization check using pre-computed length to avoid O(n) traversals
|
||||
@spec check_batch_utilization(non_neg_integer(), integer()) :: :ok
|
||||
defp check_batch_utilization(batch_length, max_size) when batch_length > max_size * 0.8 do
|
||||
Logger.warning("Batch size approaching limit",
|
||||
batch_status:
|
||||
LogSanitizer.log_data(
|
||||
current_size: length(batch),
|
||||
current_size: batch_length,
|
||||
max_size: max_size,
|
||||
utilization_percent: trunc(length(batch) / max_size * 100)
|
||||
utilization_percent: trunc(batch_length / max_size * 100)
|
||||
)
|
||||
)
|
||||
end
|
||||
|
||||
defp check_batch_utilization(_, _), do: :ok
|
||||
defp check_batch_utilization(_batch_length, _max_size), do: :ok
|
||||
|
||||
@spec process_batch(list(map())) :: :ok
|
||||
defp process_batch(packets) do
|
||||
|
|
@ -272,10 +269,10 @@ defmodule Aprsme.PacketConsumer do
|
|||
@spec process_chunk(list(map())) :: {non_neg_integer(), non_neg_integer()}
|
||||
defp process_chunk(packets) do
|
||||
# Use Stream for memory-efficient packet preparation
|
||||
# Note: truncate_datetimes_to_second is already called inside prepare_packet_for_insert
|
||||
packet_stream =
|
||||
packets
|
||||
|> Stream.map(&prepare_packet_for_insert/1)
|
||||
|> Stream.map(&truncate_datetimes_to_second/1)
|
||||
|> Stream.reject(&is_nil/1)
|
||||
|
||||
# Separate valid and invalid packets using Stream
|
||||
|
|
@ -488,19 +485,14 @@ defmodule Aprsme.PacketConsumer do
|
|||
# Detect if packet is an item or object and set appropriate fields
|
||||
defp detect_item_or_object(attrs) do
|
||||
cond do
|
||||
# Check for object data type first (takes priority over itemname)
|
||||
attrs[:data_type] == "object" or attrs["data_type"] == "object" ->
|
||||
object_name = extract_object_name(attrs)
|
||||
|
||||
attrs
|
||||
|> Map.put(:object_name, object_name)
|
||||
|> Map.put(:is_object, true)
|
||||
|> Map.delete(:itemname)
|
||||
|> Map.delete("itemname")
|
||||
|
||||
# Check information field for object format (starts with ;)
|
||||
is_binary(attrs[:information_field]) and String.starts_with?(attrs[:information_field], ";") ->
|
||||
object_name = extract_object_name_from_info_field(attrs[:information_field])
|
||||
# Check for object data type or information field starting with ;
|
||||
attrs[:data_type] == "object" or attrs["data_type"] == "object" or
|
||||
(is_binary(attrs[:information_field]) and
|
||||
String.starts_with?(attrs[:information_field], ";")) ->
|
||||
# Try data_extended first, then fall back to parsing information_field
|
||||
object_name =
|
||||
extract_object_name(attrs) ||
|
||||
extract_object_name_from_info_field(attrs[:information_field])
|
||||
|
||||
attrs
|
||||
|> Map.put(:object_name, object_name)
|
||||
|
|
@ -531,9 +523,12 @@ defmodule Aprsme.PacketConsumer do
|
|||
end
|
||||
end
|
||||
|
||||
defp extract_object_name_from_info_field(nil), do: nil
|
||||
|
||||
defp extract_object_name_from_info_field(info_field) do
|
||||
# Object format: ;OBJECTNAM*DDHHMMz...
|
||||
case Regex.run(~r/^;([A-Z0-9 ]{9})\*/, info_field) do
|
||||
# Object format: ;OBJECTNAM*DDHHMMz... or ;OBJECTNAM_DDHHMMz... (killed)
|
||||
# Object names are 9 printable ASCII characters followed by * (alive) or _ (killed)
|
||||
case Regex.run(~r/^;(.{9})[*_]/, info_field) do
|
||||
[_, name] -> String.trim(name)
|
||||
_ -> nil
|
||||
end
|
||||
|
|
|
|||
|
|
@ -37,47 +37,6 @@ defmodule Aprsme.DbOptimizerTest do
|
|||
end
|
||||
end
|
||||
|
||||
describe "copy_insert/3" do
|
||||
test "inserts rows when count is <= 1000 using regular_batch_insert" do
|
||||
now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
|
||||
|
||||
rows = [
|
||||
["test_copy_small_1", 10, now, now],
|
||||
["test_copy_small_2", 20, now, now]
|
||||
]
|
||||
|
||||
count =
|
||||
DbOptimizer.copy_insert(
|
||||
"packet_counters",
|
||||
["counter_type", "count", "inserted_at", "updated_at"],
|
||||
rows
|
||||
)
|
||||
|
||||
assert count == 2
|
||||
end
|
||||
|
||||
test "falls back to regular_batch_insert when > 1000 rows (COPY fails in sandbox)" do
|
||||
now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
|
||||
|
||||
rows =
|
||||
for i <- 1..1001 do
|
||||
["test_copy_large_#{i}", i, now, now]
|
||||
end
|
||||
|
||||
count =
|
||||
DbOptimizer.copy_insert(
|
||||
"packet_counters",
|
||||
["counter_type", "count", "inserted_at", "updated_at"],
|
||||
rows
|
||||
)
|
||||
|
||||
# COPY will fail in sandbox, falls back to regular_batch_insert
|
||||
# Some rows may conflict on unique index but on_conflict: :nothing handles that
|
||||
assert is_integer(count)
|
||||
assert count > 0
|
||||
end
|
||||
end
|
||||
|
||||
describe "optimized_batch_insert/3" do
|
||||
test "inserts valid entries and returns {count, 0}" do
|
||||
now = NaiveDateTime.truncate(NaiveDateTime.utc_now(), :second)
|
||||
|
|
|
|||
|
|
@ -1,10 +1,86 @@
|
|||
defmodule Aprsme.PacketConsumerTest do
|
||||
use Aprsme.DataCase, async: false
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Aprsme.Packet
|
||||
alias Aprsme.PacketConsumer
|
||||
alias Aprsme.StreamingPacketsPubSub
|
||||
|
||||
describe "object name extraction" do
|
||||
test "extracts object name with hyphens from information_field" do
|
||||
events = [
|
||||
%{
|
||||
sender: "db0sda",
|
||||
lat: 33.169,
|
||||
lon: -96.492,
|
||||
data_type: "object",
|
||||
destination: "APRS",
|
||||
path: "TCPIP*,qAC,SECOND",
|
||||
information_field: ";P-K5SGD *192200z3310.16NP09629.53W#PHG0-100DAPNET POCSAG Transmitter",
|
||||
raw_packet: "db0sda>APRS,TCPIP*,qAC,SECOND:;P-K5SGD *192200z3310.16NP09629.53W#"
|
||||
}
|
||||
]
|
||||
|
||||
state = %{batch: [], batch_length: 0, batch_size: 1, batch_timeout: 1000, max_batch_size: 100, timer: nil}
|
||||
|
||||
{:noreply, [], _} = PacketConsumer.handle_events(events, nil, state)
|
||||
Process.sleep(50)
|
||||
|
||||
packet = Repo.one(from p in Packet, where: p.sender == "db0sda")
|
||||
assert packet
|
||||
assert packet.object_name == "P-K5SGD"
|
||||
end
|
||||
|
||||
test "extracts object name with special characters" do
|
||||
events = [
|
||||
%{
|
||||
sender: "OBJTEST1",
|
||||
lat: 35.0,
|
||||
lon: -75.0,
|
||||
data_type: "object",
|
||||
destination: "APRS",
|
||||
path: "WIDE1-1",
|
||||
information_field: ";#146.760 *192200z3500.00N/07500.00W#Repeater",
|
||||
raw_packet: "OBJTEST1>APRS:;#146.760 *192200z3500.00N/07500.00W#Repeater"
|
||||
}
|
||||
]
|
||||
|
||||
state = %{batch: [], batch_length: 0, batch_size: 1, batch_timeout: 1000, max_batch_size: 100, timer: nil}
|
||||
|
||||
{:noreply, [], _} = PacketConsumer.handle_events(events, nil, state)
|
||||
Process.sleep(50)
|
||||
|
||||
packet = Repo.one(from p in Packet, where: p.sender == "OBJTEST1")
|
||||
assert packet
|
||||
assert packet.object_name == "#146.760"
|
||||
end
|
||||
|
||||
test "extracts object name for killed objects" do
|
||||
events = [
|
||||
%{
|
||||
sender: "OBJTEST2",
|
||||
lat: 35.0,
|
||||
lon: -75.0,
|
||||
data_type: "object",
|
||||
destination: "APRS",
|
||||
path: "WIDE1-1",
|
||||
information_field: ";P-KILLED _192200z3500.00N/07500.00W#Removed",
|
||||
raw_packet: "OBJTEST2>APRS:;P-KILLED _192200z3500.00N/07500.00W#Removed"
|
||||
}
|
||||
]
|
||||
|
||||
state = %{batch: [], batch_length: 0, batch_size: 1, batch_timeout: 1000, max_batch_size: 100, timer: nil}
|
||||
|
||||
{:noreply, [], _} = PacketConsumer.handle_events(events, nil, state)
|
||||
Process.sleep(50)
|
||||
|
||||
packet = Repo.one(from p in Packet, where: p.sender == "OBJTEST2")
|
||||
assert packet
|
||||
assert packet.object_name == "P-KILLED"
|
||||
end
|
||||
end
|
||||
|
||||
describe "handle_events/3 with stream processing" do
|
||||
test "processes packets using streams without memory accumulation" do
|
||||
# Create test packets
|
||||
|
|
@ -25,6 +101,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
# Initialize consumer state
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 100,
|
||||
batch_timeout: 1000,
|
||||
max_batch_size: 1000,
|
||||
|
|
@ -86,6 +163,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
# Set batch size to 5 to trigger immediate processing
|
||||
batch_size: 5,
|
||||
batch_timeout: 1000,
|
||||
|
|
@ -139,6 +217,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 1,
|
||||
batch_timeout: 1000,
|
||||
max_batch_size: 100,
|
||||
|
|
@ -170,6 +249,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 50,
|
||||
batch_timeout: 1000,
|
||||
# Only 100 will be processed
|
||||
|
|
@ -230,6 +310,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 2,
|
||||
batch_timeout: 1000,
|
||||
max_batch_size: 100,
|
||||
|
|
@ -264,6 +345,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 1,
|
||||
batch_timeout: 1000,
|
||||
max_batch_size: 100,
|
||||
|
|
@ -299,6 +381,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
|
||||
state = %{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 100,
|
||||
batch_timeout: 100,
|
||||
max_batch_size: 200,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue