refactor: promote hot-path GenServer state maps to structs
Converts the inline state maps in Is, PacketConsumer, LeaderElection, and CircuitBreaker into structs with @type t specs. Enforced keys catch typos on state updates and give dialyzer a concrete type to check against instead of map() in every handle_* clause. Is' nested login_params and packet_stats are extracted to their own modules (Aprsme.Is.LoginParams, Aprsme.Is.PacketStats) rather than defined inline, per the project's no-nested-modules rule.
This commit is contained in:
parent
d1409bcc7d
commit
5ef97adaf7
6 changed files with 106 additions and 69 deletions
|
|
@ -9,19 +9,27 @@ defmodule Aprsme.CircuitBreaker do
|
|||
require Logger
|
||||
|
||||
@type state :: :closed | :open | :half_open
|
||||
@type service_state :: %{
|
||||
state: state(),
|
||||
failure_count: non_neg_integer(),
|
||||
failure_threshold: non_neg_integer(),
|
||||
timeout: non_neg_integer(),
|
||||
last_failure_time: DateTime.t() | nil,
|
||||
recovery_timeout: non_neg_integer()
|
||||
}
|
||||
|
||||
@default_failure_threshold 5
|
||||
@default_timeout 5000
|
||||
@default_recovery_timeout 30_000
|
||||
|
||||
defstruct state: :closed,
|
||||
failure_count: 0,
|
||||
failure_threshold: @default_failure_threshold,
|
||||
timeout: @default_timeout,
|
||||
last_failure_time: nil,
|
||||
recovery_timeout: @default_recovery_timeout
|
||||
|
||||
@type service_state :: %__MODULE__{
|
||||
state: state(),
|
||||
failure_count: non_neg_integer(),
|
||||
failure_threshold: pos_integer(),
|
||||
timeout: pos_integer(),
|
||||
last_failure_time: DateTime.t() | nil,
|
||||
recovery_timeout: pos_integer()
|
||||
}
|
||||
|
||||
# Public API
|
||||
|
||||
@doc """
|
||||
|
|
@ -223,14 +231,7 @@ defmodule Aprsme.CircuitBreaker do
|
|||
end
|
||||
|
||||
defp default_service_state do
|
||||
%{
|
||||
state: :closed,
|
||||
failure_count: 0,
|
||||
failure_threshold: @default_failure_threshold,
|
||||
timeout: @default_timeout,
|
||||
last_failure_time: nil,
|
||||
recovery_timeout: @default_recovery_timeout
|
||||
}
|
||||
%__MODULE__{}
|
||||
end
|
||||
|
||||
defp calculate_current_state(%{
|
||||
|
|
|
|||
|
|
@ -12,6 +12,18 @@ defmodule Aprsme.Cluster.LeaderElection do
|
|||
# Maximum time to wait for cluster formation before proceeding with election (30 seconds)
|
||||
@max_cluster_wait 30_000
|
||||
|
||||
defstruct is_leader: false,
|
||||
leader_node: nil,
|
||||
cluster_enabled: false,
|
||||
election_forced: false
|
||||
|
||||
@type t :: %__MODULE__{
|
||||
is_leader: boolean(),
|
||||
leader_node: node() | nil,
|
||||
cluster_enabled: boolean(),
|
||||
election_forced: boolean()
|
||||
}
|
||||
|
||||
def start_link(opts) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
|
@ -72,7 +84,7 @@ defmodule Aprsme.Cluster.LeaderElection do
|
|||
# Initialize cached leadership state
|
||||
:persistent_term.put({__MODULE__, :is_leader}, false)
|
||||
|
||||
{:ok, %{is_leader: false, leader_node: nil, cluster_enabled: cluster_enabled, election_forced: false}}
|
||||
{:ok, %__MODULE__{cluster_enabled: cluster_enabled}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
|
|
|
|||
|
|
@ -2,26 +2,39 @@ defmodule Aprsme.Is do
|
|||
@moduledoc false
|
||||
use GenServer
|
||||
|
||||
alias Aprsme.Is.LoginParams
|
||||
alias Aprsme.Is.PacketStats
|
||||
|
||||
require Logger
|
||||
|
||||
@aprs_timeout 60 * 1000
|
||||
@keepalive_interval 20 * 1000
|
||||
@backpressure_safety_valve_timeout 30_000
|
||||
|
||||
@type state :: %{
|
||||
defstruct [
|
||||
:server,
|
||||
:port,
|
||||
:socket,
|
||||
:timer,
|
||||
:keepalive_timer,
|
||||
:connected_at,
|
||||
:packet_stats,
|
||||
:login_params,
|
||||
:safety_valve_timer,
|
||||
buffer: "",
|
||||
backpressure_active: false
|
||||
]
|
||||
|
||||
@type t :: %__MODULE__{
|
||||
server: charlist() | String.t(),
|
||||
port: pos_integer(),
|
||||
socket: :ssl.sslsocket() | nil,
|
||||
timer: reference() | nil,
|
||||
keepalive_timer: reference() | nil,
|
||||
connected_at: DateTime.t(),
|
||||
packet_stats: map(),
|
||||
connected_at: DateTime.t() | nil,
|
||||
packet_stats: PacketStats.t(),
|
||||
buffer: String.t(),
|
||||
login_params: %{
|
||||
user_id: String.t(),
|
||||
passcode: String.t(),
|
||||
filter: String.t()
|
||||
},
|
||||
login_params: LoginParams.t(),
|
||||
backpressure_active: boolean(),
|
||||
safety_valve_timer: reference() | nil
|
||||
}
|
||||
|
|
@ -66,31 +79,19 @@ defmodule Aprsme.Is do
|
|||
connected_at = DateTime.utc_now()
|
||||
|
||||
# Initialize packet statistics
|
||||
packet_stats = %{
|
||||
total_packets: 0,
|
||||
last_packet_at: nil,
|
||||
packets_per_second: 0,
|
||||
last_second_count: 0,
|
||||
last_second_timestamp: System.system_time(:second)
|
||||
}
|
||||
packet_stats = %PacketStats{last_second_timestamp: System.system_time(:second)}
|
||||
|
||||
# Initialize state without requiring immediate connection
|
||||
state = %{
|
||||
state = %__MODULE__{
|
||||
server: server,
|
||||
port: port,
|
||||
socket: nil,
|
||||
timer: nil,
|
||||
keepalive_timer: nil,
|
||||
connected_at: connected_at,
|
||||
packet_stats: packet_stats,
|
||||
buffer: "",
|
||||
login_params: %{
|
||||
login_params: %LoginParams{
|
||||
user_id: aprs_user_id,
|
||||
passcode: aprs_passcode,
|
||||
filter: default_filter
|
||||
},
|
||||
backpressure_active: false,
|
||||
safety_valve_timer: nil
|
||||
}
|
||||
}
|
||||
|
||||
# Try to connect initially, but don't fail if it doesn't work
|
||||
|
|
@ -591,13 +592,13 @@ defmodule Aprsme.Is do
|
|||
end
|
||||
end
|
||||
|
||||
@spec update_packet_stats(map(), integer()) :: map()
|
||||
@spec update_packet_stats(PacketStats.t(), integer()) :: PacketStats.t()
|
||||
defp update_packet_stats(stats, current_time) do
|
||||
new_total = stats.total_packets + 1
|
||||
|
||||
# Check if we need to reset the per-second counter
|
||||
if current_time - stats.last_second_timestamp >= 1 do
|
||||
%{
|
||||
%PacketStats{
|
||||
total_packets: new_total,
|
||||
last_packet_at: DateTime.utc_now(),
|
||||
packets_per_second: 1,
|
||||
|
|
@ -608,11 +609,11 @@ defmodule Aprsme.Is do
|
|||
new_second_count = stats.last_second_count + 1
|
||||
|
||||
%{
|
||||
total_packets: new_total,
|
||||
last_packet_at: DateTime.utc_now(),
|
||||
packets_per_second: new_second_count,
|
||||
last_second_count: new_second_count,
|
||||
last_second_timestamp: stats.last_second_timestamp
|
||||
stats
|
||||
| total_packets: new_total,
|
||||
last_packet_at: DateTime.utc_now(),
|
||||
packets_per_second: new_second_count,
|
||||
last_second_count: new_second_count
|
||||
}
|
||||
end
|
||||
end
|
||||
|
|
@ -622,15 +623,9 @@ defmodule Aprsme.Is do
|
|||
defp server_to_string(server) when is_binary(server), do: server
|
||||
defp server_to_string(server), do: to_string(server)
|
||||
|
||||
@spec default_packet_stats() :: map()
|
||||
@spec default_packet_stats() :: PacketStats.t()
|
||||
defp default_packet_stats do
|
||||
%{
|
||||
total_packets: 0,
|
||||
last_packet_at: nil,
|
||||
packets_per_second: 0,
|
||||
last_second_count: 0,
|
||||
last_second_timestamp: System.system_time(:second)
|
||||
}
|
||||
%PacketStats{last_second_timestamp: System.system_time(:second)}
|
||||
end
|
||||
|
||||
# Helper function to recursively convert structs to maps
|
||||
|
|
|
|||
10
lib/aprsme/is/login_params.ex
Normal file
10
lib/aprsme/is/login_params.ex
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
defmodule Aprsme.Is.LoginParams do
|
||||
@moduledoc false
|
||||
defstruct [:user_id, :passcode, :filter]
|
||||
|
||||
@type t :: %__MODULE__{
|
||||
user_id: String.t(),
|
||||
passcode: String.t(),
|
||||
filter: String.t()
|
||||
}
|
||||
end
|
||||
16
lib/aprsme/is/packet_stats.ex
Normal file
16
lib/aprsme/is/packet_stats.ex
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
defmodule Aprsme.Is.PacketStats do
|
||||
@moduledoc false
|
||||
defstruct total_packets: 0,
|
||||
last_packet_at: nil,
|
||||
packets_per_second: 0,
|
||||
last_second_count: 0,
|
||||
last_second_timestamp: 0
|
||||
|
||||
@type t :: %__MODULE__{
|
||||
total_packets: non_neg_integer(),
|
||||
last_packet_at: DateTime.t() | nil,
|
||||
packets_per_second: non_neg_integer(),
|
||||
last_second_count: non_neg_integer(),
|
||||
last_second_timestamp: integer()
|
||||
}
|
||||
end
|
||||
|
|
@ -11,12 +11,19 @@ defmodule Aprsme.PacketConsumer do
|
|||
|
||||
require Logger
|
||||
|
||||
@type state :: %{
|
||||
batch: list(map()),
|
||||
defstruct batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: 100,
|
||||
batch_timeout: 1000,
|
||||
max_batch_size: 1000,
|
||||
timer: nil
|
||||
|
||||
@type t :: %__MODULE__{
|
||||
batch: [map()],
|
||||
batch_length: non_neg_integer(),
|
||||
batch_size: integer(),
|
||||
batch_timeout: integer(),
|
||||
max_batch_size: integer(),
|
||||
batch_size: pos_integer(),
|
||||
batch_timeout: pos_integer(),
|
||||
max_batch_size: pos_integer(),
|
||||
timer: reference() | nil
|
||||
}
|
||||
|
||||
|
|
@ -46,7 +53,7 @@ defmodule Aprsme.PacketConsumer do
|
|||
subscribe_to = opts[:subscribe_to] || [{Aprsme.PacketProducer, max_demand: opts[:max_demand] || 250}]
|
||||
|
||||
{:consumer,
|
||||
%{
|
||||
%__MODULE__{
|
||||
batch: [],
|
||||
batch_length: 0,
|
||||
batch_size: batch_size,
|
||||
|
|
@ -110,11 +117,7 @@ defmodule Aprsme.PacketConsumer do
|
|||
)
|
||||
end
|
||||
|
||||
new_state =
|
||||
state
|
||||
|> reset_batch_timer()
|
||||
|> Map.put(:batch, carryover_batch)
|
||||
|> Map.put(:batch_length, carryover_length)
|
||||
new_state = %{reset_batch_timer(state) | batch: carryover_batch, batch_length: carryover_length}
|
||||
|
||||
{:noreply, [], new_state}
|
||||
end
|
||||
|
|
@ -135,7 +138,7 @@ defmodule Aprsme.PacketConsumer do
|
|||
end
|
||||
|
||||
# Helper functions with pattern matching
|
||||
@spec reset_batch_timer(state()) :: state()
|
||||
@spec reset_batch_timer(t()) :: t()
|
||||
defp reset_batch_timer(%{timer: nil} = state) do
|
||||
new_timer = Process.send_after(self(), :process_batch, state.batch_timeout)
|
||||
%{state | batch: [], batch_length: 0, timer: new_timer}
|
||||
|
|
@ -166,7 +169,7 @@ defmodule Aprsme.PacketConsumer do
|
|||
end
|
||||
|
||||
# Helper to start batch timer
|
||||
@spec start_batch_timer(state()) :: state()
|
||||
@spec start_batch_timer(t()) :: t()
|
||||
defp start_batch_timer(%{batch_timeout: timeout} = state) do
|
||||
timer = Process.send_after(self(), :process_batch, timeout)
|
||||
%{state | timer: timer}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue