amqpx

defmodule Amqpx.Connection do
  @moduledoc """
  Functions to operate on Connections.
  """

  import Amqpx.Core

  alias Amqpx.{Connection, DNS, Helper}

  defstruct [:pid]
  @type t :: %Connection{pid: pid}

  @default_params [
    username: "guest",
    password: "guest",
    virtual_host: "/",
    host: ~c"localhost",
    port: 5672,
    channel_max: 0,
    frame_max: 0,
    heartbeat: 10,
    connection_timeout: 50_000,
    ssl_options: :none,
    client_properties: [],
    socket_options: [],
    auth_mechanisms: [
      &:amqp_auth_mechanisms.plain/3,
      &:amqp_auth_mechanisms.amqplain/3
    ]
  ]

  @doc """
  Opens a new connection without a name.

  Behaves exactly like `open(options_or_uri, :undefined)`. See `open/2`.
  """
  @spec open(keyword | String.t()) :: {:ok, t()} | {:error, atom()} | {:error, any()}
  def open(options_or_uri \\ []) when is_binary(options_or_uri) or is_list(options_or_uri) do
    open(options_or_uri, :undefined)
  end

  @doc """
  Opens a new Connection to an Amqpx broker.

  The connections created by this module are supervised under  amqp_client's supervision tree.
  Please note that connections do not get restarted automatically by the supervision tree in
  case of a failure. If you need robust connections and channels, use monitors on the returned
  connection PID.

  This function can be called in three ways:

    * with a list of options and a name
    * with an Amqpx URI and a name
    * with an Amqpx URI and a list of options (in this case, the options are merged with
      the Amqpx URI, taking precedence on same keys)

  ## Options

    * `:username` - The name of a user registered with the broker (defaults to `"guest"`);
    * `:password` - The password of user (defaults to `"guest"`);
    * `:virtual_host` - The name of a virtual host in the broker (defaults to `"/"`);
    * `:host` - The hostname of the broker (defaults to `"localhost"`);
    * `:port` - The port the broker is listening on (defaults to `5672`);
    * `:channel_max` - The channel_max handshake parameter (defaults to `0`);
    * `:frame_max` - The frame_max handshake parameter (defaults to `0`);
    * `:heartbeat` - The hearbeat interval in seconds (defaults to `10`);
    * `:connection_timeout` - The connection timeout in milliseconds (defaults to `50000`);
    * `:ssl_options` - Enable SSL by setting the location to cert files (defaults to `:none`);
    * `:client_properties` - A list of extra client properties to be sent to the server, defaults to `[]`;
    * `:socket_options` - Extra socket options. These are appended to the default options. \
                          See http://www.erlang.org/doc/man/inet.html#setopts-2 and http://www.erlang.org/doc/man/gen_tcp.html#connect-4 \
                          for descriptions of the available options.

  ## Enabling SSL

  To enable SSL, supply the following in the `ssl_options` field:

    * `cacertfile` - Specifies the certificates of the root Certificate Authorities that we wish to implicitly trust;
    * `certfile` - The client's own certificate in PEM format;
    * `keyfile` - The client's private key in PEM format;

  ### Example

  ```
  Amqpx.Connection.open port: 5671,
                       ssl_options: [cacertfile: '/path/to/testca/cacert.pem',
                                     certfile: '/path/to/client/cert.pem',
                                     keyfile: '/path/to/client/key.pem',
                                     # only necessary with intermediate CAs
                                     # depth: 2,
                                     verify: :verify_peer,
                                     fail_if_no_peer_cert: true]
  ```

  ## Connection name

  RabbitMQ supports user-specified connection names since version 3.6.2.

  Connection names are human-readable strings that will be displayed in the management UI.
  Connection names do not have to be unique and cannot be used as connection identifiers.

  If the name is `:undefined`, the connection is not registered with any name.

  ## Examples

      iex> options = [host: "localhost", port: 5672, virtual_host: "/", username: "guest", password: "guest"]
      iex> Amqpx.Connection.open(options, :undefined)
      {:ok, %Amqpx.Connection{}}

      iex> Amqpx.Connection.open("amqp://guest:guest@localhost", port: 5673)
      {:ok, %Amqpx.Connection{}}

      iex> Amqpx.Connection.open("amqp://guest:guest@localhost", "a-connection-with-a-name")
      {:ok, %Amqpx.Connection{}}

  """
  @spec open(keyword | String.t(), String.t() | :undefined | keyword) ::
          {:ok, t()} | {:error, atom()} | {:error, any()}
  def open(options_or_uri, options_or_name)

  def open(uri, name) when is_binary(uri) and (is_binary(name) or name == :undefined) do
    open(uri, name, _options = [])
  end

  def open(options, name) when is_list(options) and (is_binary(name) or name == :undefined) do
    do_open(options, @default_params, name)
  end

  def open(uri, options) when is_binary(uri) and is_list(options) do
    open(uri, :undefined, options)
  end

  @doc """
  Opens a new connection with a name, merging the given Amqpx URI and options.

  This function opens a new connection by merging the given Amqpx URI `uri` and
  the given `options` and assigns it the name `name`.

  The options in `options` take precedence over the values in `uri`.

  See `open/2` for options.

  ## Examples

      iex> Amqpx.Connection.open("amqp://guest:guest@localhost", "a-connection-name", port: 5673)
      {:ok, %Amqpx.Connection{}}

  """
  @spec open(String.t(), String.t() | :undefined, keyword) ::
          {:ok, t()} | {:error, atom()} | {:error, any()}
  def open(uri, name, options) when is_binary(uri) and is_list(options) do
    case uri |> String.to_charlist() |> :amqp_uri.parse() do
      {:ok, amqp_params} ->
        amqp_params = amqp_params_network(amqp_params)
        do_open(options, amqp_params, name)

      error ->
        error
    end
  end

  @doc false
  @spec merge_options(keyword, keyword) :: tuple
  def merge_options(params, default) do
    default = normalize_ssl_options(default)

    amqp_params_network(
      username: keys_get(params, default, :username),
      password: Helper.get_password(params, default),
      virtual_host: keys_get(params, default, :virtual_host),
      host: params |> keys_get(default, :host) |> to_charlist(),
      port: keys_get(params, default, :port),
      channel_max: keys_get(params, default, :channel_max),
      frame_max: keys_get(params, default, :frame_max),
      heartbeat: keys_get(params, default, :heartbeat),
      connection_timeout: keys_get(params, default, :connection_timeout),
      ssl_options: keys_get(params, default, :ssl_options),
      client_properties: keys_get(params, default, :client_properties),
      socket_options: keys_get(params, default, :socket_options),
      auth_mechanisms: keys_get(params, default, :auth_mechanisms)
    )
  end

  # Gets the value from k1. If empty, gets the value from k2.
  defp keys_get(k1, k2, key) do
    Keyword.get(k1, key, Keyword.get(k2, key))
  end

  @doc """
  Closes an open Connection.
  """
  @spec close(t) :: :ok | {:error, any}
  def close(conn) do
    case :amqp_connection.close(conn.pid, 5_000) do
      :ok -> :ok
      error -> {:error, error}
    end
  end

  @spec do_open(Keyword.t(), Keyword.t(), String.t()) ::
          {:ok, t()} | {:error, atom()} | {:error, any()}
  defp do_open(params, default_params, name) do
    params
    |> keys_get(default_params, :host)
    |> to_charlist()
    |> DNS.resolve_ips()
    |> Enum.reduce_while(nil, fn ip, _ ->
      amqp_params = params |> Keyword.put(:host, ip) |> merge_options(default_params)

      case :amqp_connection.start(amqp_params, name) do
        {:ok, pid} ->
          {:halt, {:ok, %Connection{pid: pid}}}

        error ->
          {:cont, error}
      end
    end)
  end

  defp normalize_ssl_options(options) when is_list(options) do
    for {k, v} <- options do
      if k in [:cacertfile, :certfile, :keyfile] do
        {k, to_charlist(v)}
      else
        {k, v}
      end
    end
  end

  defp normalize_ssl_options(options), do: options
end