HaimKortovich
c8a58c3f58
All checks were successful
Build and Publish / build-release (push) Successful in 1m38s
88 lines
2.7 KiB
Elixir
88 lines
2.7 KiB
Elixir
defmodule PolicyService.Consumers.SolicitationTaskConsumer do
|
|
use GenServer
|
|
require Logger
|
|
|
|
alias PolicyService.CommandedApp
|
|
alias PolicyService.Commands.CarPolicy
|
|
alias PolicyService.Aggregates.PolicyId
|
|
|
|
@exchange "workload_service.events.solicitation_task_completed"
|
|
@queue "policy_service.solicitation_task_completed"
|
|
@routing_key "solicitation_task.completed"
|
|
|
|
def start_link(opts \\ []) do
|
|
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
|
end
|
|
|
|
def init(_opts) do
|
|
{:ok, conn} = AMQP.Connection.open(Application.fetch_env!(:policy_service, :amqp_url))
|
|
{:ok, channel} = AMQP.Channel.open(conn)
|
|
|
|
AMQP.Queue.declare(channel, @queue, durable: true)
|
|
AMQP.Queue.bind(channel, @queue, @exchange, routing_key: @routing_key)
|
|
{:ok, _tag} = AMQP.Basic.consume(channel, @queue)
|
|
|
|
Logger.info("SolicitationTaskConsumer started, listening on #{@queue}")
|
|
|
|
{:ok, %{channel: channel}}
|
|
end
|
|
|
|
def handle_info({:basic_deliver, payload, meta}, state) do
|
|
:ok =
|
|
case process(payload) do
|
|
:ok ->
|
|
AMQP.Basic.ack(state.channel, meta.delivery_tag)
|
|
|
|
{:error, reason} ->
|
|
Logger.error("SolicitationTaskConsumer: failed to process: #{inspect(reason)}")
|
|
AMQP.Basic.reject(state.channel, meta.delivery_tag, requeue: false)
|
|
end
|
|
|
|
{:noreply, state}
|
|
end
|
|
|
|
defp process(payload) do
|
|
with {:ok, event} <- Jason.decode(payload),
|
|
:ok <- dispatch(event) do
|
|
:ok
|
|
end
|
|
end
|
|
|
|
def handle_info({:basic_consume_ok, _}, state), do: {:noreply, state}
|
|
def handle_info({:basic_cancel, _}, state), do: {:stop, :normal, state}
|
|
def handle_info({:basic_cancel_ok, _}, state), do: {:noreply, state}
|
|
|
|
defp dispatch(%{
|
|
"application_id" => %{
|
|
"org_id" => org_id,
|
|
"policy_type" => policy_type,
|
|
"application_id" => app_id
|
|
},
|
|
"task_info" => task_info,
|
|
"submission" => submission
|
|
}) do
|
|
cmd =
|
|
case policy_type do
|
|
"car" ->
|
|
%CarPolicy.RecordPolicyIssued{
|
|
id: PolicyId.new(org_id, policy_type, app_id),
|
|
provider_policy_number: Map.get(submission, "provider_policy_number"),
|
|
effective_date: Map.get(submission, "effective_date"),
|
|
expiry_date: Map.get(submission, "expiry_date"),
|
|
issued_at:
|
|
Map.get(submission, "issued_at") || DateTime.utc_now() |> DateTime.to_iso8601()
|
|
}
|
|
end
|
|
|
|
case CommandedApp.dispatch(cmd, consistency: :strong) do
|
|
:ok ->
|
|
Logger.info("SolicitationTaskConsumer: issued policy for #{app_id}")
|
|
:ok
|
|
|
|
{:error, reason} ->
|
|
Logger.error("SolicitationTaskConsumer: failed to issue policy - #{inspect(reason)}")
|
|
{:error, reason}
|
|
end
|
|
end
|
|
end
|