Added Ecto, Cassandra and Auctions repo

lib/neon_vertigo_service.ex

@@ -8,6 +8,7 @@ defmodule NeonVertigoService do
   @impl true
   def start(_start_type, _start_args) do
     children = [
+      NeonVertigoService.Repo,
       NeonVertigoService.Mqtt.Supervisor,
     ]
 

lib/neon_vertigo_service/message_processor.ex

@@ -1,20 +1,14 @@
 defmodule NeonVertigoService.MessageProcessor do
-  @ets_auctions NeonVertigoService.Auctions
-  @ets_bids NeonVertigoService.Bids
-
-  def init_tables() do
-    @ets_auctions = :ets.new(@ets_auctions, [:named_table])
-    @ets_bids = :ets.new(@ets_bids, [:named_table])
-    :ok
-  end
-
   def process_message(["auctions", "create"], payload, _properties) do
-    gid = payload["gid"]
-    data = Map.take(payload, ["author_gid", "min_bet"]) |> Enum.map(fn {k, v} -> {String.to_atom(k), v} end) |> Map.new
-    {min_bet, _} = Decimal.parse(data[:min_bet])
-    data = Map.put(data, :min_bet, min_bet)
-    :ets.insert_new(@ets_auctions, {gid, data})
-    :noreply
+    changeset = Auction.changeset(%Auction{}, payload)
+    reply = case NeonVertigoService.Repo.insert(changeset) do
+      {:ok, auction} ->
+        %{ok: true, gid: auction.gid}
+      {:error, changeset} ->
+        %{ok: false, gid: changeset |> Ecto.Changeset.get_field(:gid)}
+    end
+
+    {:reply, reply}
   end
 
   def process_message(_, _, _) do

lib/neon_vertigo_service/mqtt/server.ex

@@ -11,8 +11,6 @@ defmodule NeonVertigoService.Mqtt.Server do
 
   # Callbacks
   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,
@@ -34,10 +32,14 @@ defmodule NeonVertigoService.Mqtt.Server do
     IO.inspect(msg)
     %{payload: payload_str, topic: topic_str, properties: properties} = msg
     topic = String.split(topic_str, "/")
+    response_topic = properties[:"Response-Topic"]
     payload = JSON.decode!(payload_str)
 
     case NeonVertigoService.MessageProcessor.process_message(topic, payload, properties) do
-      {:reply, topic, data} -> :emqtt.publish(mqtt_pid, topic, data)
+      {:reply, data} ->
+        if is_binary(response_topic) && String.length(response_topic) > 0 do
+          :emqtt.publish(mqtt_pid, response_topic, JSON.encode!(data), retain: true)
+        end
       :noreply -> nil
     end
 

lib/neon_vertigo_service/repo.ex

@@ -0,0 +1,3 @@
+defmodule NeonVertigoService.Repo do
+  use Ecto.Repo, otp_app: :neon_vertigo_service, adapter: Exandra
+end

lib/schemas/auction.ex

@@ -0,0 +1,18 @@
+defmodule Auction do
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  schema "auctions" do
+    field :gid, :string, primary_key: true
+    field :author_gid, :string
+    field :starting_price, :decimal
+  end
+
+  def changeset(auction, params \\ %{}) do
+    auction
+    |> cast(params, [:gid, :author_gid, :starting_price])
+    |> validate_required([:gid, :author_gid, :starting_price])
+    |> validate_number(:starting_price, greater_than_or_equal_to: 0)
+    |> unique_constraint([:gid])
+  end
+end