Basic routing + account creation

lib/neon_vertigo_service/mqtt/server.ex

@@ -1,4 +1,4 @@
-defmodule NeonVertigoService.Mqtt.Service do
+defmodule NeonVertigoService.Mqtt.Server do
   use GenServer
 
   def start_link(_) do
@@ -6,30 +6,41 @@ defmodule NeonVertigoService.Mqtt.Service do
   end
   
   def init(_) do
-    {:ok, {}, {:continue, :init_mqtt}}
+    {:ok, {}, {:continue, :post_init}}
   end
 
   # Callbacks
-  def handle_continue(:init_mqtt, {}) do
+  def handle_continue(:post_init, {}) do
+    :ok = NeonVertigoService.MessageProcessor.init_tables()
+
     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] ++ mqtt_config]}
+          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/test/+", [{:qos, 1}]}])
+    :emqtt.subscribe(mqtt_pid, %{}, [{"$share/neon-vertigo-service/auctions/+", [{:qos, 1}]}])
 
     {:noreply, {mqtt_pid}}
   end
 
   # Handling MQTT messages
-  def handle_info({:publish, msg}, state) do
+  def handle_info({:publish, msg}, {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, "/")
+    payload = JSON.decode!(payload_str)
+
+    case NeonVertigoService.MessageProcessor.process_message(topic, payload, properties) do
+      {:reply, topic, data} -> :emqtt.publish(mqtt_pid, topic, data)
+      :noreply -> nil
+    end
+
     {:noreply, state}
   end
 end

lib/neon_vertigo_service/mqtt/supervisor.ex

@@ -10,7 +10,7 @@ defmodule NeonVertigoService.Mqtt.Supervisor do
   @impl true
   def init(_init_arg) do
     children = [
-      NeonVertigoService.Mqtt.Service
+      NeonVertigoService.Mqtt.Server
     ]
 
     Supervisor.init(children, strategy: :one_for_all)