defmodule NeonVertigoService.MessageProcessor do alias NeonVertigoService.Repo import Ecto.Query, only: [from: 2] def process_message(mqtt_pid, {"auctions", "created"}, payload, properties) do changeset = Auction.changeset(%Auction{}, payload) reply = case 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, mqtt_pid, properties, [qos: 2]) end def process_message(mqtt_pid, {"auctions", auction_gid, "bid"}, payload, properties) do auction = Repo.get(Auction, auction_gid) if is_nil(auction) do {:not_found, auction_gid} else changeset = Bid.changeset(%Bid{}, Map.put(payload, "auction_gid", auction.gid)) reply = case Repo.insert(changeset) do {:ok, bid} -> bid_data = %{ bid: bid.bid, bidder_gid: bid.bidder_gid, timestamp: bid.bid_timestamp } send_mqtt(bid_data, mqtt_pid, "notify/new_bid/#{auction.gid}", qos: 2) %{ok: true} {:error, changeset} -> %{ok: false} end reply(reply, mqtt_pid, properties) end end def process_message(_, _, _, _) do {:ok, :unknown_message} end defp reply(data, mqtt_pid, orig_properties, message_properties \\ []) do response_topic = orig_properties[:"Response-Topic"] if is_nil(response_topic) do {:ok, :no_response_topic} else send_mqtt(data, mqtt_pid, response_topic, message_properties) end end defp send_mqtt(data, mqtt_pid, response_topic, message_properties \\ []) defp send_mqtt(data, mqtt_pid, response_topic, message_properties) when is_binary(response_topic) and byte_size(response_topic) > 0 do if String.valid?(response_topic) && String.length(response_topic) > 0 do :emqtt.publish(mqtt_pid, response_topic, JSON.encode!(data), message_properties) else {:error, :invalid_topic} end end defp send_mqtt(_, _, _, _), do: {:error, :invalid_topic} end