ryanandClaude
4a928b7067
- Add dashboard development commands (just dashboard, mix phx.server)
- Document Dashboard.Application and Dashboard.MqttSubscriber components
- Add comprehensive dashboard section with MQTT configuration details
- Include critical implementation notes for Tortoise handler return values
- Fix handle_message to return {:ok, state} instead of [] to prevent crashes
🤖 Generated with [Claude Code](https://claude.ai/code)
Co-Authored-By: Claude <noreply@anthropic.com>
97 lines
2.6 KiB
Elixir
97 lines
2.6 KiB
Elixir
defmodule Dashboard.MqttSubscriber do
|
|
@moduledoc """
|
|
Simple MQTT subscriber for development dashboard.
|
|
"""
|
|
use GenServer
|
|
require Logger
|
|
|
|
alias Phoenix.PubSub
|
|
|
|
@pubsub_topic "systant:hosts"
|
|
|
|
def start_link(opts) do
|
|
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
|
end
|
|
|
|
def get_hosts do
|
|
GenServer.call(__MODULE__, :get_hosts)
|
|
end
|
|
|
|
@impl true
|
|
def init(_opts) do
|
|
# Start MQTT connection directly with hostname-based client ID to avoid conflicts
|
|
{:ok, hostname} = :inet.gethostname()
|
|
client_id = "systant-dashboard-#{hostname}"
|
|
connection_opts = [
|
|
client_id: client_id,
|
|
server: {Tortoise.Transport.Tcp, host: "mqtt.home", port: 1883},
|
|
handler: {__MODULE__, []},
|
|
subscriptions: [{"systant/+/stats", 0}]
|
|
]
|
|
|
|
case Tortoise.Connection.start_link(connection_opts) do
|
|
{:ok, _pid} ->
|
|
Logger.info("Dashboard MQTT subscriber connected successfully")
|
|
{:ok, %{hosts: %{}}}
|
|
{:error, {:already_started, _pid}} ->
|
|
Logger.info("Dashboard MQTT connection already exists, reusing")
|
|
{:ok, %{hosts: %{}}}
|
|
{:error, reason} ->
|
|
Logger.error("Failed to connect to MQTT broker: #{inspect(reason)}")
|
|
{:stop, reason}
|
|
end
|
|
end
|
|
|
|
@impl true
|
|
def handle_call(:get_hosts, _from, state) do
|
|
{:reply, state.hosts, state}
|
|
end
|
|
|
|
@impl true
|
|
def handle_info(_msg, state) do
|
|
{:noreply, state}
|
|
end
|
|
|
|
# Tortoise handler callbacks
|
|
def connection(status, state) do
|
|
Logger.info("MQTT connection status: #{status}")
|
|
{:ok, state}
|
|
end
|
|
|
|
def subscription(status, topic, state) do
|
|
Logger.info("MQTT subscription status for #{topic}: #{status}")
|
|
{:ok, state}
|
|
end
|
|
|
|
def handle_message(topic, payload, state) do
|
|
topic_parts = if is_binary(topic), do: String.split(topic, "/"), else: topic
|
|
case topic_parts do
|
|
["systant", hostname, "stats"] ->
|
|
case Jason.decode(payload) do
|
|
{:ok, data} ->
|
|
host_data = Map.put(data, "last_seen", DateTime.utc_now())
|
|
|
|
# Broadcast to LiveView
|
|
PubSub.broadcast(Dashboard.PubSub, @pubsub_topic, {:host_update, hostname, host_data})
|
|
|
|
# Update our state
|
|
GenServer.cast(__MODULE__, {:update_host, hostname, host_data})
|
|
|
|
{:error, _reason} ->
|
|
:ok
|
|
end
|
|
_ ->
|
|
:ok
|
|
end
|
|
{:ok, state}
|
|
end
|
|
|
|
@impl true
|
|
def handle_cast({:update_host, hostname, host_data}, state) do
|
|
updated_hosts = Map.put(state.hosts, hostname, host_data)
|
|
{:noreply, %{state | hosts: updated_hosts}}
|
|
end
|
|
|
|
@impl true
|
|
def terminate(_reason, _state), do: []
|
|
end |