Current section

Files

Jump to
exq lib exq support opts.ex
Raw

lib/exq/support/opts.ex

defmodule Exq.Support.Opts do
@moduledoc """
Exq supported options.
"""
alias Exq.Support.Coercion
alias Exq.Support.Config
@doc """
Returns top supervisor's name default is `Exq.Sup`.
"""
def top_supervisor(name) do
name = name || Config.get(:name)
"#{name}.Sup" |> String.to_atom()
end
defp conform_opts(opts) do
mode = opts[:mode] || Config.get(:mode)
redis = redis_client_name(opts[:name])
opts = [{:redis, redis} | opts]
redis_opts = redis_opts(opts)
server_opts = server_opts(mode, opts)
{redis_opts, server_opts}
end
def redis_client_name(name) do
name = name || Config.get(:name)
"#{name}.Redis.Client" |> String.to_atom()
end
def redis_inspect_opts(opts \\ []) do
args = redis_opts(opts)
case args do
[url, options] -> [url, mask_password(options)]
[options] -> [mask_password(options)]
end
|> inspect()
end
def redis_opts(opts \\ []) do
redis_options = opts[:redis_options] || Config.get(:redis_options)
socket_opts = opts[:socket_opts] || Config.get(:socket_opts) || []
redis_options =
Keyword.merge(
[name: opts[:redis], socket_opts: socket_opts],
redis_options
)
if url = opts[:url] || Config.get(:url) do
[url, redis_options]
else
if Keyword.has_key?(redis_options, :sentinel) do
[redis_options]
else
host = opts[:host] || Config.get(:host)
port = Coercion.to_integer(opts[:port] || Config.get(:port))
database = Coercion.to_integer(opts[:database] || Config.get(:database))
password = opts[:password] || Config.get(:password)
[
Keyword.merge(
[host: host, port: port, database: database, password: password],
redis_options
)
]
end
end
end
@doc """
Returns `{redis_module, redis_args, gen_server_opts}`.
"""
def redis_worker_opts(opts) do
{redis_opts, opts} = conform_opts(opts)
{Redix, redis_opts, opts}
end
defp mask_password(options) do
options =
if Keyword.has_key?(options, :password) do
Keyword.update!(options, :password, fn
nil -> nil
_ -> "*****"
end)
else
options
end
options =
if Keyword.has_key?(options, :sentinel) do
Keyword.update!(options, :sentinel, &mask_password/1)
else
options
end
if Keyword.has_key?(options, :sentinels) do
Keyword.update!(options, :sentinels, fn sentinels ->
Enum.map(sentinels, &mask_password/1)
end)
else
options
end
end
defp server_opts(:default, opts) do
scheduler_enable =
Coercion.to_boolean(opts[:scheduler_enable] || Config.get(:scheduler_enable))
namespace = opts[:namespace] || Config.get(:namespace)
scheduler_poll_timeout =
Coercion.to_integer(opts[:scheduler_poll_timeout] || Config.get(:scheduler_poll_timeout))
poll_timeout = Coercion.to_integer(opts[:poll_timeout] || Config.get(:poll_timeout))
shutdown_timeout =
Coercion.to_integer(opts[:shutdown_timeout] || Config.get(:shutdown_timeout))
manager = Exq.Manager.Server.server_name(opts[:name])
enqueuer = Exq.Enqueuer.Server.server_name(opts[:name])
stats = Exq.Stats.Server.server_name(opts[:name])
scheduler = Exq.Scheduler.Server.server_name(opts[:name])
workers_sup = Exq.Worker.Supervisor.supervisor_name(opts[:name])
middleware = Exq.Middleware.Server.server_name(opts[:name])
metadata = Exq.Worker.Metadata.server_name(opts[:name])
queue_configs = opts[:queues] || Config.get(:queues)
per_queue_concurrency = cast_concurrency(opts[:concurrency] || Config.get(:concurrency))
queues = get_queues(queue_configs)
concurrency = get_concurrency(queue_configs, per_queue_concurrency)
default_middleware = Config.get(:middleware)
heartbeat_enable =
Coercion.to_boolean(Keyword.get(opts, :heartbeat_enable, Config.get(:heartbeat_enable)))
heartbeat_interval =
Coercion.to_integer(opts[:heartbeat_interval] || Config.get(:heartbeat_interval))
missed_heartbeats_allowed =
Coercion.to_integer(
opts[:missed_heartbeats_allowed] || Config.get(:missed_heartbeats_allowed)
)
[
scheduler_enable: scheduler_enable,
namespace: namespace,
scheduler_poll_timeout: scheduler_poll_timeout,
workers_sup: workers_sup,
poll_timeout: poll_timeout,
enqueuer: enqueuer,
metadata: metadata,
stats: stats,
name: opts[:name],
manager: manager,
scheduler: scheduler,
queues: queues,
redis: opts[:redis],
concurrency: concurrency,
middleware: middleware,
default_middleware: default_middleware,
mode: :default,
shutdown_timeout: shutdown_timeout,
heartbeat_enable: heartbeat_enable,
heartbeat_interval: heartbeat_interval,
missed_heartbeats_allowed: missed_heartbeats_allowed
]
end
defp server_opts(mode, opts) do
namespace = opts[:namespace] || Config.get(:namespace)
[name: opts[:name], namespace: namespace, redis: opts[:redis], mode: mode]
end
defp get_queues(queue_configs) do
Enum.map(queue_configs, fn queue_config ->
case queue_config do
{queue, _concurrency} -> queue
queue -> queue
end
end)
end
defp get_concurrency(queue_configs, per_queue_concurrency) do
Enum.map(queue_configs, fn queue_config ->
case queue_config do
{queue, concurrency} -> {queue, cast_concurrency(concurrency)}
queue -> {queue, per_queue_concurrency}
end
end)
end
def cast_concurrency({module, options}), do: {module, options}
def cast_concurrency(:infinity), do: {Exq.Dequeue.Local, [concurrency: :infinity]}
def cast_concurrency(:infinite), do: {Exq.Dequeue.Local, [concurrency: :infinity]}
def cast_concurrency(x) when is_integer(x), do: {Exq.Dequeue.Local, [concurrency: x]}
def cast_concurrency(x) when is_binary(x) do
case x |> String.trim() |> String.downcase() do
"infinity" -> {Exq.Dequeue.Local, [concurrency: :infinity]}
"infinite" -> {Exq.Dequeue.Local, [concurrency: :infinity]}
x -> {Exq.Dequeue.Local, [concurrency: Coercion.to_integer(x)]}
end
end
end