defmodule MQ.Topology.Setup do @moduledoc """ Use to perform a one-off setup of exchanges and queues for `dev` and `test` environments. """ alias AMQP.{ Channel, Connection, Exchange, Queue } alias MQ.AMQPConfig alias MQ.Topology.Config alias MQ.Topology.Config.{DeadLetterQueueConfig, QueueConfig, TopicExchangeConfig} require Logger defmodule Result do defstruct channel: nil, exchanges: [], errors: [] end @spec run() :: {:ok, any()} def run do topology = Config.gen() amqp_url = AMQPConfig.url() with {:ok, connection} <- Connection.open(amqp_url), {:ok, channel} <- Channel.open(connection), results <- setup(channel, topology), :ok <- Channel.close(channel) do {:ok, results} end end defp setup(channel, topology) do topology |> Task.async_stream(&setup_exchange(channel, &1)) |> Enum.to_list() end defp setup_exchange(channel, %TopicExchangeConfig{ name: exchange, durable: durable, queues: queues }) do channel |> assert_exchange(:topic, exchange, durable) |> setup_queues(queues) end defp setup_queues(channel, queues) when is_list(queues) do Enum.map(queues, &setup_queues(channel, &1)) end defp setup_queues( channel, {%DeadLetterQueueConfig{} = dead_letter_queue, %QueueConfig{} = queue} ) do channel |> assert_dead_letter_queue(dead_letter_queue) |> assert_queue(queue) |> bind_queue(queue) end defp assert_exchange(channel, :topic, name, durable) do :ok = Exchange.topic(channel, name, durable: durable) channel end defp assert_dead_letter_queue(channel, %DeadLetterQueueConfig{name: name, durable: durable}) do {:ok, _} = Queue.declare(channel, name, durable: durable) channel end defp assert_queue(channel, %QueueConfig{ name: name, durable: durable, exclusive: exclusive, args: arguments }) do {:ok, %{queue: queue}} = Queue.declare(channel, name, durable: durable, exclusive: exclusive, arguments: arguments) {channel, queue} end defp bind_queue({channel, queue}, %QueueConfig{binding: {exchange, "", routing_key}} = config) do :ok = Queue.bind(channel, queue, exchange, routing_key: routing_key) result(queue, config) end defp bind_queue( {channel, queue}, %QueueConfig{binding: {exchange, queue, routing_key}} = config ) do :ok = Queue.bind(channel, queue, exchange, routing_key: routing_key) result(queue, config) end defp result(queue, %QueueConfig{binding: {exchange, _queue, routing_key}} = config) do result = config |> Map.take([:durable, :exclusive, :args]) |> Map.merge(%{ exchange: exchange, queue: queue, routing_key: routing_key }) Logger.debug("Declared #{queue} queue: #{inspect(result)}") result end end