defmodule MQ.ConnectionManager do alias AMQP.{Channel, Confirm, Connection} alias Core.ExponentialBackoff alias MQ.{AMQPConfig, ChannelRegistry} require Logger use GenServer @this_module __MODULE__ defmodule State do @enforce_keys [:amqp_url] defstruct amqp_url: nil, connection: nil end @spec start_link(list()) :: GenServer.on_start() def start_link(opts) do amqp_url = opts |> Keyword.get(:amqp_url, AMQPConfig.url()) # `name: @this_module` makes it callable via the module name. GenServer.start_link(@this_module, %State{amqp_url: amqp_url}, name: @this_module) end @spec request_channel(atom()) :: {:ok, Channel.t()} | {:error, any()} def request_channel(server_name) when is_atom(server_name) do GenServer.call(@this_module, {:request_channel, server_name, :normal}) end @doc """ See https://hexdocs.pm/amqp/AMQP.Confirm.html for more details. """ @spec request_confirm_channel(atom()) :: {:ok, Channel.t()} | {:error, any()} def request_confirm_channel(server_name) when is_atom(server_name) do GenServer.call(@this_module, {:request_channel, server_name, :confirm}) end @spec close_channel(atom()) :: :ok | {:error, any()} def close_channel(server_name) when is_atom(server_name) do GenServer.call(@this_module, {:close_channel, server_name}) end @impl true def init(%State{amqp_url: amqp_url} = initial_state) do :ok = ChannelRegistry.init() {:ok, %Connection{} = connection} = connect(amqp_url) {:ok, %{initial_state | connection: connection}} end @impl true def handle_call( {:request_channel, _server_name, _channel_type}, _from, %State{connection: nil} = state ), do: {:reply, {:error, :no_connection}, state} @impl true def handle_call( {:request_channel, server_name, channel_type}, _from, %State{connection: %Connection{} = connection} = state ) do server_name |> ChannelRegistry.lookup() |> case do {:ok, %Channel{}} = reply -> {:reply, reply, state} {:error, :channel_not_found} -> {:reply, open_channel(connection, server_name, channel_type), state} end end @impl true def handle_call({:close_channel, server_name}, _from, %State{} = state) do server_name |> ChannelRegistry.lookup() |> case do {:ok, %Channel{} = channel} -> {:reply, close_and_remove_channel(channel, server_name), state} error -> {:reply, error, state} end end @impl true def handle_info({:DOWN, _, :process, _pid, reason}, %State{amqp_url: amqp_url} = state) do Logger.error("Connection to #{amqp_url} lost due to #{inspect(reason)}.") # Stop GenServer. Will be restarted by Supervisor. {:stop, {:connection_lost, reason}, %{state | connection: nil}} end defp connect(amqp_url) do ExponentialBackoff.with_backoff(fn -> case Connection.open(amqp_url) do {:ok, %Connection{} = connection} = reply -> Logger.debug("Connected to #{amqp_url}.") _ = monitor_connection(connection) reply error -> Logger.error("Error connecting to #{amqp_url}: #{inspect(error)}. Retrying...") error end end) end defp open_channel(%Connection{} = connection, server_name, :confirm) do Logger.debug("Opening a confirm channel for #{server_name}.") with {:ok, channel} <- Channel.open(connection), :ok <- Confirm.select(channel), :ok <- ChannelRegistry.insert(server_name, channel) do {:ok, channel} end end defp open_channel(%Connection{} = connection, server_name, :normal) do Logger.debug("Opening a channel for #{server_name}.") with {:ok, channel} <- Channel.open(connection), :ok <- ChannelRegistry.insert(server_name, channel) do {:ok, channel} end end defp close_and_remove_channel(%Channel{} = channel, server_name) do channel |> Channel.close() |> case do :ok -> server_name |> ChannelRegistry.delete() error -> error end end # We will get notified when the connection is down # and exit the process cleanly. # # See how we handle `{:DOWN, _, :process, _pid, reason}`. defp monitor_connection(%Connection{pid: pid}), do: Process.monitor(pid) end