47 lines
1.3 KiB
Elixir
47 lines
1.3 KiB
Elixir
defmodule NeonVertigoService.Mqtt.Server do
|
|
use GenServer
|
|
|
|
alias __MODULE__, as: MqttServer
|
|
|
|
defstruct [:mqtt_pid]
|
|
|
|
def start_link(_) do
|
|
GenServer.start_link(__MODULE__, [], name: __MODULE__)
|
|
end
|
|
|
|
def init(_) do
|
|
{:ok, %MqttServer{}, {:continue, :post_init}}
|
|
end
|
|
|
|
# == Callbacks ==
|
|
|
|
def handle_continue(:post_init, %MqttServer{} = state) do
|
|
mqtt_config = Application.fetch_env!(:neon_vertigo_service, :mqtt_params)
|
|
{:ok, mqtt_pid} = Supervisor.start_child(
|
|
NeonVertigoService.Mqtt.Supervisor,
|
|
%{
|
|
id: :neon_vertigo_emqtt,
|
|
start: {:emqtt, :start_link, [[owner: self(), name: :neon_vertigo_service_emqtt, proto_ver: :v5] ++ mqtt_config]}
|
|
}
|
|
)
|
|
{:ok, _mqtt_props} = :emqtt.connect(mqtt_pid)
|
|
|
|
:emqtt.subscribe(mqtt_pid, %{}, [{"$share/neon-vertigo-service/auctions/#", [{:qos, 1}]}])
|
|
|
|
{:noreply, %{state | mqtt_pid: mqtt_pid}}
|
|
end
|
|
|
|
# Handling MQTT messages
|
|
def handle_info({:publish, msg}, %{mqtt_pid: mqtt_pid} = state) do
|
|
IO.puts("Received message")
|
|
IO.inspect(msg)
|
|
%{payload: payload_str, topic: topic_str, properties: properties} = msg
|
|
topic = String.split(topic_str, "/") |> List.to_tuple()
|
|
payload = JSON.decode!(payload_str)
|
|
|
|
NeonVertigoService.MessageProcessor.process_message(mqtt_pid, topic, payload, properties)
|
|
|
|
{:noreply, state}
|
|
end
|
|
end
|