off_broadway_mqtt_connector

defmodule OffBroadway.MQTT.Config do
  @default_transport :tcp
  @default_host "localhost"
  @default_port 1883
  @default_dequeue_interval 5000
  @default_client_id_prefix "obmp"
  @default_supervisor OffBroadway.MQTT.QueueSupervisor
  @default_registry OffBroadway.MQTT.QueueRegistry
  @default_telemetry_prefix :off_broadway_mqtt

  @moduledoc """
  Defines a data structure for configuring this library.

  ## Config options

    * `dequeue_interval` - The interval used by the producer to timout polls to
      the queue process. Defaults to `#{inspect(@default_dequeue_interval)}`.
    * `client_id_prefix` - The value is used to prefix the randomly generated
      client ids by the MQTT client. Defaults to
      #{inspect(@default_client_id_prefix)}. Keep in mind that some brokers
      limit the size  of client_id size to 23 bytes!
    * `server_opts` - See the "Server options" section for details.
    * `telemetry_prefix` - Sets the prefix for any telemery events. Defaults to
      `#{inspect(@default_telemetry_prefix)}`.

  ### Server options

  All options given with the `server_opts` option are passed to `Tortoise` when
  starting the connection process. The following options can be given:

    * `host` - The host the MQTT client uses by default. Defaults to
      `#{inspect(@default_host)}`.
    * `port` - The port the MQTT client uses by default. Defaults to
      `#{inspect(@default_port)}`.
    * `transport` - The protocol the MQTT client uses by default. Defaults to
      `#{inspect(@default_transport)}`.
    * See `Tortoise.Connection.start_link/1` for further options.

  ## Dependency injection

  Besides the other config options it is also possible to replace some modules
  used by providing an alternative implementation:

    * `queue_supervisor` - The supervisor that supervises the queue processes
      started by the producer. Defaults to `#{inspect(@default_supervisor)}`.
    * `queue_registry` - The registry that is used to register the queue
      processes started by the producer. Defaults to
      `#{inspect(@default_registry)}`.
    * `acknowledger` - The `Broadway.Acknowledger` implementation used when
      building the message strucs.
    * `client` - The MQTT client module.
    * `handler`- The handler module used by the default client.
    * `queue` - The queue used in the handler and producer to enqueue / dequeue
    messages.


  ## Compiletime configuration

  The following options must be given at compile time:

  * `telemetry_enabled` - Enables telemetry events if set to true. This option
    is disabled by default.

  ## Configuration example

      use Mix.Config

      config :off_broadway_mqtt,
        client_id_prefix: "sensor_data_processor",
        server_opts: [
          host: "vernemq",
          port: 8883,
          transport: :ssl
        ],
        handler: MyApp.BetterHandler

  ## Building configurations

      iex> OffBroadway.MQTT.Config.new()
      %OffBroadway.MQTT.Config{
        acknowledger: OffBroadway.MQTT.Acknowledger,
        client: OffBroadway.MQTT.Client,
        client_id_prefix: "obmp",
        dequeue_interval: 5000,
        handler: OffBroadway.MQTT.Handler,
        producer: OffBroadway.MQTT.Producer,
        queue: OffBroadway.MQTT.Queue,
        queue_registry: OffBroadway.MQTT.QueueRegistry,
        queue_supervisor: OffBroadway.MQTT.QueueSupervisor,
        server: {:tcp, [host: 'localhost', port: 1883]},
        telemetry_prefix: :off_broadway_mqtt
      }

      iex> OffBroadway.MQTT.Config.new(
      ...>   telemetry_prefix: :test,
      ...>     server_opts: [host: "vernemq", port: 8883, transport: :ssl]
      ...>   )
      %OffBroadway.MQTT.Config{
        acknowledger: OffBroadway.MQTT.Acknowledger,
        client: OffBroadway.MQTT.Client,
        client_id_prefix: "obmp",
        dequeue_interval: 5000,
        handler: OffBroadway.MQTT.Handler,
        producer: OffBroadway.MQTT.Producer,
        queue: OffBroadway.MQTT.Queue,
        queue_registry: OffBroadway.MQTT.QueueRegistry,
        queue_supervisor: OffBroadway.MQTT.QueueSupervisor,
        server: {:ssl, [host: 'vernemq', port: 8883]},
        telemetry_prefix: :test
      }

  Keep in mind that any option with nil or the empty string will be removed from
  the server options to prevent issues when configuring the application from
  environment variables.
  """

  alias OffBroadway.MQTT.Acknowledger
  alias OffBroadway.MQTT.Client
  alias OffBroadway.MQTT.Handler
  alias OffBroadway.MQTT.Producer
  alias OffBroadway.MQTT.Queue

  @type transport :: :tcp | :ssl

  @type raw_server_opt ::
          {:host, String.t()}
          | {:port, non_neg_integer | String.t()}
          | {:transport, transport, String.t()}
          | {atom, any}

  @type raw_server_opts :: [raw_server_opt, ...]

  @type server_opt ::
          {:host, String.t()}
          | {:port, non_neg_integer}
          | {:transport, transport}
          | {atom, any}
  @type server_opts :: [server_opt, ...]

  @type server :: {transport, server_opts}

  @type option ::
          {:acknowledger, module}
          | {:client, module}
          | {:client_id_prefix, String.t()}
          | {:dequeue_interval, non_neg_integer}
          | {:handler, module}
          | {:producer, module}
          | {:queue, module}
          | {:queue_registry, GenServer.name()}
          | {:queue_supervisor, GenServer.name()}
          | {:server_opts, raw_server_opts}
          | {:telemetry_prefix, atom}
          | {atom, any}

  @type options :: [option]

  @type t :: %__MODULE__{
          acknowledger: module,
          client: module,
          client_id_prefix: String.t(),
          dequeue_interval: non_neg_integer,
          handler: module,
          producer: module,
          queue: module,
          queue_registry: GenServer.name(),
          queue_supervisor: GenServer.name(),
          server: server,
          telemetry_prefix: atom
        }

  defstruct [
    :acknowledger,
    :client,
    :client_id_prefix,
    :dequeue_interval,
    :handler,
    :producer,
    :queue,
    :queue_supervisor,
    :queue_registry,
    :server,
    :telemetry_prefix
  ]

  @doc """
  Returns a `t:t/0` with values from the applicatoin config.

  Use this function if you want to build a config based on the configured
  defaults for the application.
  """
  @spec new_from_app_config(options) :: t
  def new_from_app_config(overrides \\ []) do
    :off_broadway_mqtt
    |> Application.get_all_env()
    |> Keyword.merge(overrides)
    |> new()
  end

  @doc """
  Builds a `t:t/0` from the given keyword list.

  Use this function if you want to build a config only from the passed options.
  Any application config is ignored if using this function.
  """
  @spec new(options) :: t
  def new(opts \\ []) when is_list(opts) do
    {transport, server_opts} =
      opts
      |> Keyword.get(:server_opts, [])
      |> Keyword.update(:host, @default_host, &parse_host/1)
      |> Keyword.update(:port, @default_port, &parse_port/1)
      |> Keyword.update(:transport, @default_transport, &parse_transport/1)
      |> Keyword.pop(:transport, @default_transport)

    server_opts = Enum.reject(server_opts, &is_empty?/1)

    struct_opts =
      opts
      |> Keyword.put_new(:acknowledger, Acknowledger)
      |> Keyword.put_new(:client, Client)
      |> Keyword.put_new(:client_id_prefix, @default_client_id_prefix)
      |> Keyword.put_new(:dequeue_interval, @default_dequeue_interval)
      |> Keyword.put_new(:handler, Handler)
      |> Keyword.put_new(:producer, Producer)
      |> Keyword.put_new(:queue, Queue)
      |> Keyword.put_new(:queue_registry, @default_registry)
      |> Keyword.put_new(:queue_supervisor, @default_supervisor)
      |> Keyword.put_new(:server, {transport, server_opts})
      |> Keyword.put_new(:telemetry_prefix, @default_telemetry_prefix)

    struct(__MODULE__, struct_opts)
  end

  defp is_empty?({_, ""}), do: true
  defp is_empty?({_, nil}), do: true
  defp is_empty?(_), do: false

  defp parse_host(host) when is_binary(host), do: host

  defp parse_port(port) when is_integer(port), do: port

  defp parse_port(port) when is_binary(port) do
    case Integer.parse(port) do
      {port, ""} ->
        port

      _ ->
        raise ArgumentError,
              "invalid port configured for #{__MODULE__}: \"#{port}\""
    end
  end

  defp parse_port(port) do
    raise ArgumentError,
          "invalid port configured for #{__MODULE__}: \"#{port}\""
  end

  defp parse_transport(t) when t in ["TCP", "tcp", :tcp], do: :tcp
  defp parse_transport(t) when t in ["SSL", "ssl", :ssl], do: :ssl

  defp parse_transport(t) do
    raise ArgumentError,
          "invalid transport configured for #{__MODULE__}: \"#{t}\".\n" <>
            "Allowed values are `TCP` and `SSL`."
  end
end