From 480cc084cf70059453dd0daee6b1fb54e31e5cfe Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Mon, 4 May 2026 09:52:34 -0500 Subject: [PATCH] refactor(pskr): drop emqtt, talk MQTT 3.1.1 over :gen_tcp directly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit emqtt's transitive `quicer` dep needs CMake + msquic submodule clones to build, which the slim Debian builder image doesn't ship — the prod CI build was failing inside `mix deps.compile`. Tortoise311 is the obvious pure-Elixir alternative but it bundles a gen_state_machine + retry/inflight stack we don't use because we're subscribe-only at QoS 0. Implementing the wire protocol inline is ~150 lines in `Microwaveprop.Pskr.Mqtt` (CONNECT/SUBSCRIBE/PINGREQ/DISCONNECT builders, CONNACK/PUBLISH/SUBACK/PINGRESP parser, varint codec). Pskr.Client now uses :gen_tcp with `active: :once` framing, buffers partial reads, schedules its own keepalive, and parses inbound packets in a tight loop. Drops `BUILD_WITHOUT_QUIC=1` from the Dockerfile — no longer relevant since there's no native dep at all in the build chain now. --- Dockerfile | 8 - lib/microwaveprop/pskr/client.ex | 246 +++++++++++++++++--------- lib/microwaveprop/pskr/mqtt.ex | 182 +++++++++++++++++++ mix.exs | 13 +- mix.lock | 6 - test/microwaveprop/pskr/mqtt_test.exs | 103 +++++++++++ 6 files changed, 454 insertions(+), 104 deletions(-) create mode 100644 lib/microwaveprop/pskr/mqtt.ex create mode 100644 test/microwaveprop/pskr/mqtt_test.exs diff --git a/Dockerfile b/Dockerfile index f01c9573..2455c528 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,14 +43,6 @@ RUN mix local.hex --force \ # set build ENV ENV MIX_ENV="prod" -# Skip the QUIC native build inside emqtt's transitive `quicer` dep. -# We only need plain TCP MQTT (port 1883) for the PSK Reporter -# firehose, and quicer's CMake-driven build pulls in extra -# toolchain requirements (cmake, OpenSSL headers) that aren't worth -# adding to the slim build image just to leave the support -# unused. -ENV BUILD_WITHOUT_QUIC="1" - # install mix dependencies (changes only when mix.exs/mix.lock change) COPY mix.exs mix.lock ./ COPY vendor vendor diff --git a/lib/microwaveprop/pskr/client.ex b/lib/microwaveprop/pskr/client.ex index 670a4c0c..61511c79 100644 --- a/lib/microwaveprop/pskr/client.ex +++ b/lib/microwaveprop/pskr/client.ex @@ -1,41 +1,48 @@ defmodule Microwaveprop.Pskr.Client do @moduledoc """ - MQTT client for the PSK Reporter live spot firehose. + MQTT 3.1.1 client for the PSK Reporter live spot firehose. - Connects to `mqtt.pskreporter.info:1883` (plain TCP — the broker - doesn't expose TLS), subscribes to a curated set of topics, and - forwards every payload to `Pskr.Aggregator.ingest/2`. Each - topic anchors on USA (DXCC 291) on either the sender or receiver - side so the broker pre-filters out the bulk of the firehose - before it ever reaches us — the alternative would be subscribing - to `pskr/filter/v2/#` and dropping ~95% client-side. + Connects to `mqtt.pskreporter.info:1883` over plain TCP (the + broker doesn't expose TLS), subscribes to a curated set of + topics, and forwards every PUBLISH payload to + `Pskr.Aggregator.ingest/2`. Each topic anchors on USA (DXCC 291) + on either the sender or receiver side so the broker pre-filters + out the bulk of the firehose before it reaches us — the + alternative would be subscribing to `pskr/filter/v2/#` and + dropping ~95% client-side. + + ## No external MQTT library + + We talk to the broker with `:gen_tcp` and the codec in + `Microwaveprop.Pskr.Mqtt`. emqtt's transitive `quicer` dep was a + CMake/native-build mess in our slim Debian builder image; the + alternative pure-Elixir client (tortoise311) bundles a + gen_state_machine + retry/inflight stack we don't use because + we're subscribe-only at QoS 0. Implementing the wire protocol + inline is ~150 lines of binary parsing — less code than the + dependency-management overhead of either choice. ## Cluster singleton All replicas start this GenServer; **only one across the cluster actually connects to MQTT**. We elect a leader via - `:global.register_name/2` — atomic across the BEAM cluster - (libcluster gives us connectivity). The losing nodes stay in - `:standby` mode: no MQTT connection, no DB writes, just a - cheap idle process watching for the leader to disappear. + `:global.register_name/2` (atomic across the BEAM cluster, which + libcluster gives us). The losing nodes stay in `:standby` mode: + no socket, no DB writes, just an idle process watching for the + leader's `nodedown` to re-run election. - When the leader's node goes down (deploy, OOM, network - partition), `:global` evicts its registration cluster-wide and - emits `{:nodedown, node}` to every subscriber. Standby clients - race to re-register; one wins and starts a fresh connection - inside `handle_continue/2`. The handover loses at most one flush - cycle (~60 s) of in-flight aggregates from the old leader's - Aggregator — acceptable, since the upsert-on-conflict logic - merges any overlapping path-hours additively. + Handover loses at most one flush cycle (~60 s) of in-flight + aggregates from the old leader's Aggregator — acceptable, since + the upsert-on-conflict logic in `Pskr.Aggregator` merges any + overlapping path-hours additively. ## Bands Subscribes to **VHF and up** by default (6m, 2m, 70cm, 23cm). HF spots are dominated by ionospheric propagation, which isn't what - this project models — we deliberately skip HF rather than waste - bandwidth. Microwave bands are listed too but spots there are - vanishingly rare on PSK Reporter (FT8 activity is mostly - 144 MHz / 432 MHz / 1296 MHz). + this project models. Microwave bands are listed too but spots + there are vanishingly rare (FT8 activity is mostly 144 MHz / + 432 MHz / 1296 MHz). ## Off in dev/test @@ -46,11 +53,14 @@ defmodule Microwaveprop.Pskr.Client do use GenServer alias Microwaveprop.Pskr.Aggregator + alias Microwaveprop.Pskr.Mqtt require Logger @broker_host ~c"mqtt.pskreporter.info" @broker_port 1883 + @keepalive_seconds 60 + @keepalive_ms @keepalive_seconds * 1000 # USA = 291 (continental). PSK Reporter splits Alaska (6) and # Hawaii (110) as separate DXCCs but neither is currently in the @@ -68,22 +78,20 @@ defmodule Microwaveprop.Pskr.Client do def init(opts) do Process.flag(:trap_exit, true) # `:net_kernel.monitor_nodes/1` lands `{:nodeup, node}` / - # `{:nodedown, node}` messages in our mailbox — that's what - # tells a standby client the leader vanished and election - # should re-run. + # `{:nodedown, node}` in our mailbox — that's what tells a + # standby client the leader vanished and election should + # re-run. :ok = :net_kernel.monitor_nodes(true) - bands = Keyword.get(opts, :bands, @default_bands) - aggregator = Keyword.get(opts, :aggregator, Aggregator) - client_id = Keyword.get(opts, :client_id, default_client_id()) - state = %{ role: :standby, - pid: nil, - bands: bands, - aggregator: aggregator, - client_id: client_id, - received: 0 + socket: nil, + buffer: <<>>, + bands: Keyword.get(opts, :bands, @default_bands), + aggregator: Keyword.get(opts, :aggregator, Aggregator), + client_id: Keyword.get(opts, :client_id, default_client_id()), + received: 0, + ping_timer: nil } {:ok, state, {:continue, :elect}} @@ -97,17 +105,15 @@ defmodule Microwaveprop.Pskr.Client do {:noreply, %{state | role: :active}, {:continue, :connect}} :no -> - leader = :global.whereis_name(__MODULE__) - Logger.info("Pskr.Client standby on #{node()} (leader: #{inspect(leader)})") + Logger.info("Pskr.Client standby on #{node()} (leader: #{inspect(:global.whereis_name(__MODULE__))})") {:noreply, state} end end def handle_continue(:connect, state) do - case connect_and_subscribe(state) do - {:ok, pid} -> - Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands") - {:noreply, %{state | pid: pid}} + case do_connect(state) do + {:ok, socket, ping_timer} -> + {:noreply, %{state | socket: socket, ping_timer: ping_timer}} {:error, reason} -> Logger.error("Pskr.Client connect failed: #{inspect(reason)} — retrying in 30s") @@ -119,76 +125,150 @@ defmodule Microwaveprop.Pskr.Client do @impl true def handle_info(:reconnect, state), do: {:noreply, state, {:continue, :connect}} - # When any node joins or leaves, standby clients re-attempt - # leader election. The active leader is unaffected — its - # registration is still valid until *its* node drops out. + def handle_info(:ping, %{socket: socket} = state) when is_port(socket) do + case :gen_tcp.send(socket, Mqtt.pingreq()) do + :ok -> + {:noreply, %{state | ping_timer: schedule_ping()}} + + {:error, reason} -> + Logger.warning("Pskr.Client PINGREQ send failed: #{inspect(reason)}") + {:noreply, drop_socket(state), {:continue, :connect}} + end + end + + def handle_info(:ping, state), do: {:noreply, state} + + # Standby clients re-run election on any cluster topology change. + # Active leaders ignore — their registration stays valid until + # *their* node disappears. def handle_info({event, _node}, %{role: :standby} = state) when event in [:nodeup, :nodedown] do {:noreply, state, {:continue, :elect}} end def handle_info({event, _node}, state) when event in [:nodeup, :nodedown], do: {:noreply, state} - # `:emqtt` delivers PUBLISH frames to the owner process as - # `{:publish, %{topic: …, payload: …, qos: …}}`. - def handle_info({:publish, %{payload: payload}}, state) do - Aggregator.ingest(state.aggregator, payload) - {:noreply, %{state | received: state.received + 1}} + def handle_info({:tcp, socket, data}, %{socket: socket} = state) do + :inet.setopts(socket, active: :once) + {:noreply, process_buffer(%{state | buffer: state.buffer <> data})} end - def handle_info({:disconnected, reason_code, _props}, state) do - Logger.warning("Pskr.Client disconnected: #{inspect(reason_code)} (received #{state.received} so far)") + def handle_info({:tcp_closed, socket}, %{socket: socket} = state) do + Logger.warning("Pskr.Client TCP closed (received #{state.received} so far) — reconnecting in 5s") Process.send_after(self(), :reconnect, 5_000) - {:noreply, %{state | pid: nil}} + {:noreply, drop_socket(state)} end - def handle_info({:EXIT, pid, reason}, %{pid: pid} = state) do - Logger.warning("Pskr.Client emqtt exited: #{inspect(reason)} — reconnecting in 5s") + def handle_info({:tcp_error, socket, reason}, %{socket: socket} = state) do + Logger.warning("Pskr.Client TCP error #{inspect(reason)} — reconnecting in 5s") Process.send_after(self(), :reconnect, 5_000) - {:noreply, %{state | pid: nil}} + {:noreply, drop_socket(state)} end def handle_info(_other, state), do: {:noreply, state} - defp connect_and_subscribe(state) do - opts = [ - host: @broker_host, - port: @broker_port, - clientid: state.client_id, - clean_start: true, - keepalive: 60, - owner: self() - ] + @impl true + def terminate(_reason, %{socket: socket}) when is_port(socket) do + _ = :gen_tcp.send(socket, Mqtt.disconnect()) + _ = :gen_tcp.close(socket) + :ok + end - with {:ok, pid} <- :emqtt.start_link(opts), - {:ok, _props} <- :emqtt.connect(pid), - :ok <- subscribe_all(pid, state.bands) do - {:ok, pid} + def terminate(_reason, _state), do: :ok + + # ---------- Connection setup ---------- + + defp do_connect(state) do + tcp_opts = [:binary, active: :once, packet: :raw, keepalive: true] + + with {:ok, socket} <- :gen_tcp.connect(@broker_host, @broker_port, tcp_opts, 10_000), + :ok <- :gen_tcp.send(socket, Mqtt.connect(state.client_id, @keepalive_seconds)), + {:ok, :connected, rest} <- await_connack(socket, <<>>), + :ok <- :gen_tcp.send(socket, subscribe_packet(state.bands)) do + Logger.info("Pskr.Client connected as #{state.client_id}, subscribed to #{length(state.bands)} bands") + # `rest` holds anything the broker sent piggybacked after + # CONNACK — feed it back into the buffer so we don't lose + # the SUBACK or an early PUBLISH. + send(self(), {:tcp, socket, rest}) + {:ok, socket, schedule_ping()} end end - defp subscribe_all(pid, bands) do + # CONNACK arrives synchronously right after CONNECT, so a small + # blocking read here keeps the state machine simple. If we don't + # get it within 10s, treat the connection as dead. + defp await_connack(socket, buffer) do + case Mqtt.parse(buffer) do + {:ok, {:connack, 0, _}, rest} -> + {:ok, :connected, rest} + + {:ok, {:connack, code, _}, _} -> + {:error, {:connack_refused, code}} + + {:incomplete, buf} -> + case :gen_tcp.recv(socket, 0, 10_000) do + {:ok, chunk} -> await_connack(socket, buf <> chunk) + {:error, _} = err -> err + end + + {:error, _} = err -> + err + end + end + + defp subscribe_packet(bands) do topics = Enum.flat_map(bands, fn band -> - [ - {topic_for_band(band, :sender_us), 0}, - {topic_for_band(band, :receiver_us), 0} - ] + [{topic_for(band, :sender_us), 0}, {topic_for(band, :receiver_us), 0}] end) - case :emqtt.subscribe(pid, %{}, topics) do - {:ok, _props, _reasons} -> :ok - other -> {:error, {:subscribe_failed, other}} - end + Mqtt.subscribe(1, topics) end # Topic order (from the broker's landing page): # pskr/filter/v2//////// - # Anchor either sender or receiver country on USA so the broker - # delivers only the half of the firehose that touches our HRRR - # domain. - defp topic_for_band(band, :sender_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/+" + defp topic_for(band, :sender_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/#{@us_dxcc}/+" + defp topic_for(band, :receiver_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/+/#{@us_dxcc}" - defp topic_for_band(band, :receiver_us), do: "pskr/filter/v2/#{band}/+/+/+/+/+/+/#{@us_dxcc}" + # ---------- Inbound dispatch ---------- + + defp process_buffer(state) do + case Mqtt.parse(state.buffer) do + {:ok, {:publish, _topic, payload}, rest} -> + Aggregator.ingest(state.aggregator, payload) + process_buffer(%{state | buffer: rest, received: state.received + 1}) + + {:ok, {:suback, _id, results}, rest} -> + if Enum.any?(results, &(&1 == 0x80)) do + Logger.warning("Pskr.Client SUBACK contains failures: #{inspect(results)}") + end + + process_buffer(%{state | buffer: rest}) + + {:ok, :pingresp, rest} -> + process_buffer(%{state | buffer: rest}) + + {:ok, _other, rest} -> + process_buffer(%{state | buffer: rest}) + + {:incomplete, buffer} -> + %{state | buffer: buffer} + + {:error, reason} -> + Logger.error("Pskr.Client parse error: #{inspect(reason)} — dropping connection") + Process.send_after(self(), :reconnect, 1_000) + drop_socket(state) + end + end + + defp drop_socket(state) do + if state.ping_timer, do: Process.cancel_timer(state.ping_timer) + if is_port(state.socket), do: :gen_tcp.close(state.socket) + %{state | socket: nil, buffer: <<>>, ping_timer: nil} + end + + defp schedule_ping, do: Process.send_after(self(), :ping, @keepalive_ms) + + # ---------- Client ID ---------- defp default_client_id do suffix = 4 |> :crypto.strong_rand_bytes() |> Base.encode16(case: :lower) diff --git a/lib/microwaveprop/pskr/mqtt.ex b/lib/microwaveprop/pskr/mqtt.ex new file mode 100644 index 00000000..ce122e03 --- /dev/null +++ b/lib/microwaveprop/pskr/mqtt.ex @@ -0,0 +1,182 @@ +defmodule Microwaveprop.Pskr.Mqtt do + @moduledoc """ + Minimal MQTT 3.1.1 codec — just the packet types this project + actually exchanges with `mqtt.pskreporter.info:1883`. + + We're a subscribe-only QoS 0 client, so the wire protocol reduces + to: + + * outbound: CONNECT, SUBSCRIBE, PINGREQ, DISCONNECT + * inbound: CONNACK, PUBLISH (QoS 0), SUBACK, PINGRESP + + No retained messages, no will, no QoS 1/2 inflight tracking, no + packet ID reuse logic. Anything beyond that is intentionally not + implemented — adding a dep just to throw away 80% of its surface + area felt worse than the ~150 lines below. + + ## Frame layout + + Every MQTT packet is `<>` where `vbi` is a variable-byte integer (1-4 bytes, + high bit = "more bytes follow"). `parse/1` splits a TCP buffer + into the next decoded packet plus any trailing bytes; on a short + read it returns `{:incomplete, buffer}` so the caller can + re-invoke after the next chunk arrives. + + See: https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html + """ + + import Bitwise + + @type packet :: + {:connack, return_code :: 0..5, session_present :: boolean()} + | {:publish, topic :: String.t(), payload :: binary()} + | {:suback, packet_id :: non_neg_integer(), qos_results :: [0..2 | 0x80]} + | :pingresp + + # ---------- Outbound builders ---------- + + @doc """ + CONNECT packet for an MQTT 3.1.1 client. Flags are hardcoded to + `clean_session=1` and no will / no auth — matches the public + pskreporter broker which doesn't require credentials. + """ + @spec connect(String.t(), pos_integer()) :: binary() + def connect(client_id, keepalive_seconds) when is_binary(client_id) do + body = << + # Variable header + 4::16, + "MQTT", + # Protocol level: 3.1.1 + 0x04, + # Connect flags: clean_session + 0x02, + keepalive_seconds::16, + # Payload: client ID + byte_size(client_id)::16, + client_id::binary + >> + + fixed_header(0x10, body) + end + + @doc """ + SUBSCRIBE packet. Topics is a list of `{topic_filter, qos}` pairs. + Returns `{packet_binary, packet_id}` — the caller tracks the + packet ID for matching the SUBACK on the wire. + """ + @spec subscribe(non_neg_integer(), [{String.t(), 0..2}]) :: binary() + def subscribe(packet_id, topics) when is_list(topics) and topics != [] do + body = + IO.iodata_to_binary([ + <> | Enum.map(topics, fn {filter, qos} -> <> end) + ]) + + # Bits 3-0 of the SUBSCRIBE first byte are reserved as 0010. + fixed_header(0x82, body) + end + + @doc "PINGREQ keep-alive packet." + @spec pingreq() :: binary() + def pingreq, do: <<0xC0, 0x00>> + + @doc "DISCONNECT packet for clean shutdown." + @spec disconnect() :: binary() + def disconnect, do: <<0xE0, 0x00>> + + # ---------- Inbound parser ---------- + + @doc """ + Parse the next packet from a buffer of received TCP bytes. + + * `{:ok, packet, rest}` — packet decoded, `rest` is the + unconsumed tail (may contain more packets). + * `{:incomplete, buffer}` — short read; caller should append + the next TCP chunk and call again. + * `{:error, reason}` — protocol violation; caller should drop + the connection. + """ + @spec parse(binary()) :: {:ok, packet(), binary()} | {:incomplete, binary()} | {:error, term()} + def parse(<<>>), do: {:incomplete, <<>>} + + def parse(<> = buffer) do + case decode_varint(rest) do + {:ok, len, after_len} when byte_size(after_len) >= len -> + <> = after_len + decode_packet(type_flags, body, tail) + + {:ok, _len, _} -> + {:incomplete, buffer} + + :incomplete -> + {:incomplete, buffer} + + {:error, _} = err -> + err + end + end + + defp decode_packet(0x20, <>, tail) do + {:ok, {:connack, return_code, (flags &&& 1) == 1}, tail} + end + + # PUBLISH: top nibble is 3, low nibble is DUP/QoS/RETAIN. We only + # subscribe at QoS 0, so the broker won't send us QoS 1/2 — but + # we still tolerate the QoS 0 form (low nibble all zero or with + # RETAIN set). Anything else is rejected so a misconfigured + # subscription surfaces loudly instead of silently dropping bytes. + defp decode_packet(type_flags, body, tail) when (type_flags &&& 0xF0) == 0x30 do + qos = type_flags >>> 1 &&& 0b11 + + case qos do + 0 -> + <> = body + {:ok, {:publish, topic, payload}, tail} + + _ -> + {:error, {:unsupported_qos, qos}} + end + end + + defp decode_packet(0x90, <>, tail) do + {:ok, {:suback, packet_id, :erlang.binary_to_list(qos_results)}, tail} + end + + defp decode_packet(0xD0, <<>>, tail), do: {:ok, :pingresp, tail} + + defp decode_packet(type_flags, _body, _tail) do + {:error, {:unhandled_packet_type, type_flags}} + end + + # ---------- Variable-byte integer ---------- + + @doc false + @spec encode_varint(non_neg_integer()) :: binary() + def encode_varint(n) when n in 0..127, do: <> + + def encode_varint(n) when n <= 268_435_455 do + <<(n &&& 0x7F) ||| 0x80>> <> encode_varint(n >>> 7) + end + + @doc false + @spec decode_varint(binary()) :: {:ok, non_neg_integer(), binary()} | :incomplete | {:error, term()} + def decode_varint(buffer), do: decode_varint(buffer, 0, 0) + + defp decode_varint(_buffer, _shift, count) when count >= 4, do: {:error, :varint_too_long} + defp decode_varint(<<>>, _shift, _count), do: :incomplete + + defp decode_varint(<<0::1, n::7, rest::binary>>, shift, _count) do + {:ok, n <<< shift, rest} + end + + defp decode_varint(<<1::1, n::7, rest::binary>>, shift, count) do + case decode_varint(rest, shift + 7, count + 1) do + {:ok, tail, remaining} -> {:ok, n <<< shift ||| tail, remaining} + other -> other + end + end + + defp fixed_header(byte1, body) do + <> <> encode_varint(byte_size(body)) <> body + end +end diff --git a/mix.exs b/mix.exs index 958d6e31..79fd9ba6 100644 --- a/mix.exs +++ b/mix.exs @@ -113,13 +113,12 @@ defmodule Microwaveprop.MixProject do # PG2 adapter (in-process). Wires up via application.ex's # PubSub child spec, not a config block. {:phoenix_pubsub_redis, "~> 3.0"}, - # MQTT client for the PSK Reporter live spot firehose - # (mqtt.pskreporter.info:1883). Used by `Microwaveprop.Pskr.Client` - # to ingest geolocated, timestamped reception reports — the - # ground-truth signal we correlate against HRRR atmospheric - # state. Disabled in dev/test by default; enabled in prod via - # `PSKR_MQTT_ENABLED=true`. - {:emqtt, "~> 1.13"}, + # MQTT 3.1.1 transport for `Microwaveprop.Pskr.Client` is + # implemented inline in `Microwaveprop.Pskr.Mqtt` — the wire + # protocol for a subscribe-only QoS 0 client is ~150 lines of + # binary parsing, less code than the dep-management overhead + # of either emqtt (transitive quicer + cmake) or tortoise311 + # (gen_state_machine + retry/inflight machinery we don't use). {:ecto_psql_extras, "~> 0.8"}, {:dialyxir, "~> 1.4", only: [:dev, :test], runtime: false}, {:live_table, "~> 0.4.1"}, diff --git a/mix.lock b/mix.lock index 03edd47a..19a88199 100644 --- a/mix.lock +++ b/mix.lock @@ -8,7 +8,6 @@ "circular_buffer": {:hex, :circular_buffer, "1.0.0", "25c004da0cba7bd8bc1bdabded4f9a902d095e20600fd15faf1f2ffbaea18a07", [:mix], [], "hexpm", "c829ec31c13c7bafd1f546677263dff5bfb006e929f25635878ac3cfba8749e5"}, "comeonin": {:hex, :comeonin, "5.5.1", "5113e5f3800799787de08a6e0db307133850e635d34e9fab23c70b6501669510", [:mix], [], "hexpm", "65aac8f19938145377cee73973f192c5645873dcf550a8a6b18187d17c13ccdb"}, "complex": {:hex, :complex, "0.6.0", "b0130086a7a8c33574d293b2e0e250f4685580418eac52a5658a4bd148f3ccf1", [:mix], [], "hexpm", "0a5fa95580dcaf30fcd60fe1aaf24327c0fe401e98c24d892e172e79498269f9"}, - "cowlib": {:hex, :cowlib, "2.16.0", "54592074ebbbb92ee4746c8a8846e5605052f29309d3a873468d76cdf932076f", [:make, :rebar3], [], "hexpm", "7f478d80d66b747344f0ea7708c187645cfcc08b11aa424632f78e25bf05db51"}, "credo": {:hex, :credo, "1.7.18", "5c5596bf7aedf9c8c227f13272ac499fe8eae6237bd326f2f07dfc173786f042", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a189d164685fd945809e862fe76a7420c4398fa288d76257662aecb909d6b3e5"}, "db_connection": {:hex, :db_connection, "2.10.0", "8ff756471e41765bd5563b633f73e9a94bbc138816e8644bb17d0d91bf260a95", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "02cdd01b45efb1b550e68edbbea41be32de9b24bb07e1ea0e9cbc522ac377e54"}, "decimal": {:hex, :decimal, "2.3.0", "3ad6255aa77b4a3c4f818171b12d237500e63525c2fd056699967a3e7ea20f62", [:mix], [], "hexpm", "a4d66355cb29cb47c3cf30e71329e58361cfcb37c34235ef3bf1d7bf3773aeac"}, @@ -17,7 +16,6 @@ "ecto_psql_extras": {:hex, :ecto_psql_extras, "0.8.8", "aa02529c97f69aed5722899f5dc6360128735a92dd169f23c5d50b1f7fdede08", [:mix], [{:ecto_sql, "~> 3.7", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:postgrex, "> 0.16.0", [hex: :postgrex, repo: "hexpm", optional: false]}, {:table_rex, "~> 3.1.1 or ~> 4.0", [hex: :table_rex, repo: "hexpm", optional: false]}], "hexpm", "04c63d92b141723ad6fed2e60a4b461ca00b3594d16df47bbc48f1f4534f2c49"}, "ecto_sql": {:hex, :ecto_sql, "3.13.5", "2f8282b2ad97bf0f0d3217ea0a6fff320ead9e2f8770f810141189d182dc304e", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "aa36751f4e6a2b56ae79efb0e088042e010ff4935fc8684e74c23b1f49e25fdc"}, "elixir_make": {:hex, :elixir_make, "0.9.0", "6484b3cd8c0cee58f09f05ecaf1a140a8c97670671a6a0e7ab4dc326c3109726", [:mix], [], "hexpm", "db23d4fd8b757462ad02f8aa73431a426fe6671c80b200d9710caf3d1dd0ffdb"}, - "emqtt": {:hex, :emqtt, "1.15.0", "20b75d79ea625ce31564f31af5e686e7b12252f05ae6680118587de0ce1cdc0e", [:rebar3], [{:cowlib, "~> 2.13", [hex: :cowlib, repo: "hexpm", optional: false]}, {:getopt, "1.0.3", [hex: :getopt, repo: "hexpm", optional: false]}, {:gun, "~> 2.1", [hex: :gun, repo: "hexpm", optional: false]}, {:quicer, "0.2.15", [hex: :quicer, repo: "hexpm", optional: false]}], "hexpm", "b1bc715a83eeb043407cc776deedeff2514b9ddf5b936937c0a04a0b1d05fe1e"}, "erlex": {:hex, :erlex, "0.2.8", "cd8116f20f3c0afe376d1e8d1f0ae2452337729f68be016ea544a72f767d9c12", [:mix], [], "hexpm", "9d66ff9fedf69e49dc3fd12831e12a8a37b76f8651dd21cd45fcf5561a8a7590"}, "esbuild": {:hex, :esbuild, "0.10.0", "b0aa3388a1c23e727c5a3e7427c932d89ee791746b0081bbe56103e9ef3d291f", [:mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "468489cda427b974a7cc9f03ace55368a83e1a7be12fba7e30969af78e5f8c70"}, "exla": {:hex, :exla, "0.11.0", "1428de9edcb297480a64611d3a72fcefe13c93c115bba6d38e910583c37e38c8", [:make, :mix], [{:elixir_make, "~> 0.6", [hex: :elixir_make, repo: "hexpm", optional: false]}, {:fine, "~> 0.1", [hex: :fine, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:nx, "~> 0.11.0", [hex: :nx, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:xla, "~> 0.10.0", [hex: :xla, repo: "hexpm", optional: false]}], "hexpm", "1067207c802bd6f28cded6a2664979ee2e25dddce95cb84be3f0a3ebfbab2c74"}, @@ -25,9 +23,7 @@ "finch": {:hex, :finch, "0.21.0", "b1c3b2d48af02d0c66d2a9ebfb5622be5c5ecd62937cf79a88a7f98d48a8290c", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.6.2 or ~> 1.7", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 1.1", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "87dc6e169794cb2570f75841a19da99cfde834249568f2a5b121b809588a4377"}, "fine": {:hex, :fine, "0.1.6", "4bf7151493443c454aac9f2fa2f34f5fefd0346a83fb5586a016c4a135c63247", [:mix], [], "hexpm", "5638eb4495488e885ebec167fa57973e5c35e1a50c344eb7666c90ec1c4e3b12"}, "gen_smtp": {:hex, :gen_smtp, "1.3.0", "62c3d91f0dcf6ce9db71bcb6881d7ad0d1d834c7f38c13fa8e952f4104a8442e", [:rebar3], [{:ranch, ">= 1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "0b73fbf069864ecbce02fe653b16d3f35fd889d0fdd4e14527675565c39d84e6"}, - "getopt": {:hex, :getopt, "1.0.3", "4f3320c1f6f26b2bec0f6c6446b943eb927a1e6428ea279a1c6c534906ee79f1", [:rebar3], [], "hexpm", "7e01de90ac540f21494ff72792b1e3162d399966ebbfc674b4ce52cb8f49324f"}, "glob_ex": {:hex, :glob_ex, "0.1.11", "cb50d3f1ef53f6ca04d6252c7fde09fd7a1cf63387714fe96f340a1349e62c93", [:mix], [], "hexpm", "342729363056e3145e61766b416769984c329e4378f1d558b63e341020525de4"}, - "gun": {:hex, :gun, "2.2.0", "b8f6b7d417e277d4c2b0dc3c07dfdf892447b087f1cc1caff9c0f556b884e33d", [:make, :rebar3], [{:cowlib, ">= 2.15.0 and < 3.0.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "76022700c64287feb4df93a1795cff6741b83fb37415c40c34c38d2a4645261a"}, "heroicons": {:git, "https://github.com/tailwindlabs/heroicons.git", "0435d4ca364a608cc75e2f8683d374e55abbae26", [tag: "v2.2.0", sparse: "optimized", depth: 1]}, "hpax": {:hex, :hpax, "1.0.3", "ed67ef51ad4df91e75cc6a1494f851850c0bd98ebc0be6e81b026e765ee535aa", [:mix], [], "hexpm", "8eab6e1cfa8d5918c2ce4ba43588e894af35dbd8e91e6e55c817bca5847df34a"}, "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"}, @@ -64,12 +60,10 @@ "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"}, "postgrex": {:hex, :postgrex, "0.22.0", "fb027b58b6eab1f6de5396a2abcdaaeb168f9ed4eccbb594e6ac393b02078cbd", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "a68c4261e299597909e03e6f8ff5a13876f5caadaddd0d23af0d0a61afcc5d84"}, "prom_ex": {:hex, :prom_ex, "1.11.0", "1f6d67f2dead92224cb4f59beb3e4d319257c5728d9638b4a5e8ceb51a4f9c7e", [:mix], [{:absinthe, ">= 1.7.0", [hex: :absinthe, repo: "hexpm", optional: true]}, {:broadway, ">= 1.1.0", [hex: :broadway, repo: "hexpm", optional: true]}, {:ecto, ">= 3.11.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:finch, "~> 0.18", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, ">= 2.10.0", [hex: :oban, repo: "hexpm", optional: true]}, {:octo_fetch, "~> 0.4", [hex: :octo_fetch, repo: "hexpm", optional: false]}, {:peep, "~> 3.0", [hex: :peep, repo: "hexpm", optional: false]}, {:phoenix, ">= 1.7.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, ">= 0.20.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, ">= 1.16.0", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, ">= 2.6.0", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:telemetry, ">= 1.0.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 1.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}, {:telemetry_metrics_prometheus_core, "~> 1.2", [hex: :telemetry_metrics_prometheus_core, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 1.1", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "76b074bc3730f0802978a7eb5c7091a65473eaaf07e99ec9e933138dcc327805"}, - "quicer": {:hex, :quicer, "0.2.15", "775fcfa09c9ce5a4a00d23c1c8b1e81cd361ebc847e43bb766efac4373cdb1b8", [:rebar3], [{:snabbkaffe, "1.0.10", [hex: :snabbkaffe, repo: "hexpm", optional: false]}], "hexpm", "52236d9384a541341706bcae438e81b209fa21d2045d3df317f8ba5beda6a8e5"}, "ranch": {:hex, :ranch, "2.2.0", "25528f82bc8d7c6152c57666ca99ec716510fe0925cb188172f41ce93117b1b0", [:make, :rebar3], [], "hexpm", "fa0b99a1780c80218a4197a59ea8d3bdae32fbff7e88527d7d8a4787eff4f8e7"}, "redix": {:hex, :redix, "1.5.3", "4eaae29c75e3285c0ff9957046b7c209aa7f72a023a17f0a9ea51c2a50ab5b0f", [:mix], [{:castore, "~> 0.1.0 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:nimble_options, "~> 0.5.0 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7b06fb5246373af41f5826b03334dfa3f636347d4d5d98b4d455b699d425ae7e"}, "req": {:hex, :req, "0.5.17", "0096ddd5b0ed6f576a03dde4b158a0c727215b15d2795e59e0916c6971066ede", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.17", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 2.0.6 or ~> 2.1", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "0b8bc6ffdfebbc07968e59d3ff96d52f2202d0536f10fef4dc11dc02a2a43e39"}, "rewrite": {:hex, :rewrite, "1.3.0", "67448ba7975690b35ba7e7f35717efcce317dbd5963cb0577aa7325c1923121a", [:mix], [{:glob_ex, "~> 0.1", [hex: :glob_ex, repo: "hexpm", optional: false]}, {:sourceror, "~> 1.0", [hex: :sourceror, repo: "hexpm", optional: false]}, {:text_diff, "~> 0.1", [hex: :text_diff, repo: "hexpm", optional: false]}], "hexpm", "d111ac7ff3a58a802ef4f193bbd1831e00a9c57b33276e5068e8390a212714a5"}, - "snabbkaffe": {:hex, :snabbkaffe, "1.0.10", "9be2f54f61fc6862391b666b2b5b76c3fa53598e2989a17cef1b48cf347a8a63", [:rebar3], [], "hexpm", "70a98df36ae756908d55b5770891d443d63c903833e3e87d544036e13d4fac26"}, "sourceror": {:hex, :sourceror, "1.12.0", "da354c5f35aad3cc1132f5d5b0d8437d865e2661c263260480bab51b5eedb437", [:mix], [], "hexpm", "755703683bd014ebcd5de9acc24b68fb874a660a568d1d63f8f98cd8a6ef9cd0"}, "spitfire": {:hex, :spitfire, "0.3.11", "79dfcb033762470de472c1c26ea2b4e3aca74700c685dbffd9a13466272c323d", [:mix], [], "hexpm", "eb6e2dadf63214e8bfe65ca9788cef2b03b01027365d78d3c0e3d9ebd3d5b7b4"}, "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"}, diff --git a/test/microwaveprop/pskr/mqtt_test.exs b/test/microwaveprop/pskr/mqtt_test.exs new file mode 100644 index 00000000..4181d939 --- /dev/null +++ b/test/microwaveprop/pskr/mqtt_test.exs @@ -0,0 +1,103 @@ +defmodule Microwaveprop.Pskr.MqttTest do + use ExUnit.Case, async: true + + alias Microwaveprop.Pskr.Mqtt + + describe "encode_varint/1 + decode_varint/1" do + # Boundary values from the MQTT 3.1.1 spec, table 2.4. + @cases [ + {0, 1}, + {127, 1}, + {128, 2}, + {16_383, 2}, + {16_384, 3}, + {2_097_151, 3}, + {2_097_152, 4}, + {268_435_455, 4} + ] + + test "round-trip encodes to the expected number of bytes" do + for {n, len} <- @cases do + encoded = Mqtt.encode_varint(n) + assert byte_size(encoded) == len, "len mismatch for #{n}" + assert {:ok, ^n, <<>>} = Mqtt.decode_varint(encoded) + end + end + + test "decode returns :incomplete on truncated input" do + # 0x80 has the continuation bit set but no following byte. + assert :incomplete == Mqtt.decode_varint(<<0x80>>) + end + + test "decode returns the unconsumed tail" do + assert {:ok, 200, "tail"} = Mqtt.decode_varint(Mqtt.encode_varint(200) <> "tail") + end + end + + describe "connect/2" do + test "produces a CONNECT packet for the given client id and keepalive" do + packet = Mqtt.connect("test-client", 60) + + # Fixed header: type=1, flags=0 → 0x10 + assert <<0x10, _rest::binary>> = packet + + # Body should contain the protocol header verbatim. + assert :binary.match(packet, "MQTT") != :nomatch + assert :binary.match(packet, "test-client") != :nomatch + end + end + + describe "subscribe/2" do + test "encodes packet id and each topic with its QoS" do + packet = Mqtt.subscribe(42, [{"foo/bar", 0}, {"baz", 1}]) + + # Type=8, reserved bits=0010 → 0x82 + assert <<0x82, _rest::binary>> = packet + assert :binary.match(packet, "foo/bar") != :nomatch + assert :binary.match(packet, "baz") != :nomatch + # Packet ID 42 lands in the variable header right after the + # remaining-length byte (which is < 128 → single byte). + assert <<0x82, _len, 0::8, 42::8, _::binary>> = packet + end + end + + describe "parse/1" do + test "decodes CONNACK with session_present + return_code" do + assert {:ok, {:connack, 0, false}, ""} = Mqtt.parse(<<0x20, 0x02, 0x00, 0x00>>) + assert {:ok, {:connack, 5, true}, ""} = Mqtt.parse(<<0x20, 0x02, 0x01, 0x05>>) + end + + test "decodes a QoS 0 PUBLISH" do + topic = "pskr/filter/v2/2m/FT8" + payload = ~s({"sq":1}) + body = <> + packet = <<0x30, byte_size(body)>> <> body + + assert {:ok, {:publish, ^topic, ^payload}, ""} = Mqtt.parse(packet) + end + + test "decodes SUBACK with per-topic QoS results" do + assert {:ok, {:suback, 7, [0, 1, 0x80]}, ""} = Mqtt.parse(<<0x90, 0x05, 0::8, 7::8, 0, 1, 0x80>>) + end + + test "decodes PINGRESP" do + assert {:ok, :pingresp, ""} = Mqtt.parse(<<0xD0, 0x00>>) + end + + test "returns :incomplete on a short remaining-length read" do + # Just the type byte, no length yet. + assert {:incomplete, <<0x30>>} = Mqtt.parse(<<0x30>>) + end + + test "returns :incomplete when body is shorter than declared length" do + # Says length=10 but only gives 3 bytes. + assert {:incomplete, _} = Mqtt.parse(<<0x30, 10, "abc">>) + end + + test "returns trailing bytes after a complete packet" do + first = <<0xD0, 0x00>> + tail = <<0xC0, 0x00>> + assert {:ok, :pingresp, ^tail} = Mqtt.parse(first <> tail) + end + end +end