16bf7540e38e434e6ed4fadc1f40894960dd12db-PR-500

Committed 07 Aug 2025 11:50AM UTC coverage: 90.805% (-1.6%) from 92.382%

Build # 16bf7540e38e434e6ed4fadc1f40894960dd12db-PR-500

Build Type

Pull #500

github

Committed by

ananthakumaran

Commit Message

Run coveralls on one build only

Pull Request Pull Request #500: Add ability to snooze job

Run Details

15 of 15 new or added lines in 2 files covered. (100.0%)

18 existing lines in 13 files now uncovered.

1195 of 1316 relevant lines covered (90.81%)

706.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.72

/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

1	defmodule Exq.Support.Opts do
2	@moduledoc """
3	Exq supported options.
4	"""
5
6	alias Exq.Support.Coercion
7	alias Exq.Support.Config
8
9	@doc """
10	Returns top supervisor's name default is `Exq.Sup`.
11	"""
12	def top_supervisor(name) do
13	name = name \|\| Config.get(:name)	127✔
14	"#{name}.Sup" \|> String.to_atom()	127✔
15	end
16
17	defp conform_opts(opts) do
18	mode = opts[:mode] \|\| Config.get(:mode)	134✔
19	redis = redis_client_name(opts[:name])	134✔
20	opts = [{:redis, redis} \| opts]	134✔
21
22	redis_opts = redis_opts(opts)	134✔
23	server_opts = server_opts(mode, opts)	134✔
24	{redis_opts, server_opts}
25	end
26
27	def redis_client_name(name) do
28	name = name \|\| Config.get(:name)	134✔
29	"#{name}.Redis.Client" \|> String.to_atom()	134✔
30	end
31
32	def redis_inspect_opts(opts \\ []) do
33	args = redis_opts(opts)	8✔
34
35	case args do
36	[url, options] -> [url, mask_password(options)]	5✔
37	[options] -> [mask_password(options)]	3✔
38	end
39	\|> inspect()	8✔
40	end
41
42	def redis_opts(opts \\ []) do
43	redis_options = opts[:redis_options] \|\| Config.get(:redis_options)	152✔
44	socket_opts = opts[:socket_opts] \|\| Config.get(:socket_opts) \|\| []	152✔
45
46	redis_options =	152✔
47	Keyword.merge(
48	[name: opts[:redis], socket_opts: socket_opts],
49	redis_options
50	)
51
52	if url = opts[:url] \|\| Config.get(:url) do	152✔
53	[url, redis_options]
54	else
55	if Keyword.has_key?(redis_options, :sentinel) do	143✔
56	[redis_options]
57	else
58	host = opts[:host] \|\| Config.get(:host)	142✔
59	port = Coercion.to_integer(opts[:port] \|\| Config.get(:port))	142✔
60	database = Coercion.to_integer(opts[:database] \|\| Config.get(:database))	141✔
61	password = opts[:password] \|\| Config.get(:password)	141✔
62
63	[
64	Keyword.merge(
65	[host: host, port: port, database: database, password: password],
66	redis_options
67	)
68	]
69	end
70	end
71	end
72
73	@doc """
74	Returns `{redis_module, redis_args, gen_server_opts}`.
75	"""
76	def redis_worker_opts(opts) do
77	{redis_opts, opts} = conform_opts(opts)	134✔
78	{Redix, redis_opts, opts}	134✔
79	end
80
81	defp mask_password(options) do
82	options =	16✔
83	if Keyword.has_key?(options, :password) do
84	Keyword.update!(options, :password, fn	7✔
85	nil -> nil	2✔
86	_ -> "*****"	5✔
87	end)
88	else
89	options	9✔
90	end
91
92	options =	16✔
93	if Keyword.has_key?(options, :sentinel) do
94	Keyword.update!(options, :sentinel, &mask_password/1)	4✔
95	else
96	options	12✔
97	end
98
99	if Keyword.has_key?(options, :sentinels) do	16✔
100	Keyword.update!(options, :sentinels, fn sentinels ->	4✔
101	Enum.map(sentinels, &mask_password/1)	4✔
102	end)
103	else
104	options	12✔
105	end
106	end
107
108	defp server_opts(:default, opts) do
109	scheduler_enable =	127✔
110	Coercion.to_boolean(opts[:scheduler_enable] \|\| Config.get(:scheduler_enable))	127✔
111
112	namespace = opts[:namespace] \|\| Config.get(:namespace)	127✔
113
114	scheduler_poll_timeout =	127✔
115	Coercion.to_integer(opts[:scheduler_poll_timeout] \|\| Config.get(:scheduler_poll_timeout))	127✔
116
117	poll_timeout = Coercion.to_integer(opts[:poll_timeout] \|\| Config.get(:poll_timeout))	127✔
118
119	shutdown_timeout =	127✔
120	Coercion.to_integer(opts[:shutdown_timeout] \|\| Config.get(:shutdown_timeout))	127✔
121
122	manager = Exq.Manager.Server.server_name(opts[:name])	127✔
123	enqueuer = Exq.Enqueuer.Server.server_name(opts[:name])	127✔
124	stats = Exq.Stats.Server.server_name(opts[:name])	127✔
125	scheduler = Exq.Scheduler.Server.server_name(opts[:name])	127✔
126	workers_sup = Exq.Worker.Supervisor.supervisor_name(opts[:name])	127✔
127	middleware = Exq.Middleware.Server.server_name(opts[:name])	127✔
128	metadata = Exq.Worker.Metadata.server_name(opts[:name])	127✔
129
130	queue_configs = opts[:queues] \|\| Config.get(:queues)	127✔
131	per_queue_concurrency = cast_concurrency(opts[:concurrency] \|\| Config.get(:concurrency))	127✔
132	queues = get_queues(queue_configs)	127✔
133	concurrency = get_concurrency(queue_configs, per_queue_concurrency)	127✔
134	default_middleware = Config.get(:middleware)	127✔
135
136	heartbeat_enable =	127✔
137	Coercion.to_boolean(Keyword.get(opts, :heartbeat_enable, Config.get(:heartbeat_enable)))
138
139	heartbeat_interval =	127✔
140	Coercion.to_integer(opts[:heartbeat_interval] \|\| Config.get(:heartbeat_interval))	127✔
141
142	missed_heartbeats_allowed =	127✔
143	Coercion.to_integer(
144	opts[:missed_heartbeats_allowed] \|\| Config.get(:missed_heartbeats_allowed)	127✔
145	)
146
147	[
148	scheduler_enable: scheduler_enable,
149	namespace: namespace,
150	scheduler_poll_timeout: scheduler_poll_timeout,
151	workers_sup: workers_sup,
152	poll_timeout: poll_timeout,
153	enqueuer: enqueuer,
154	metadata: metadata,
155	stats: stats,
156	name: opts[:name],
157	manager: manager,
158	scheduler: scheduler,
159	queues: queues,
160	redis: opts[:redis],
161	concurrency: concurrency,
162	middleware: middleware,
163	default_middleware: default_middleware,
164	mode: :default,
165	shutdown_timeout: shutdown_timeout,
166	heartbeat_enable: heartbeat_enable,
167	heartbeat_interval: heartbeat_interval,
168	missed_heartbeats_allowed: missed_heartbeats_allowed
169	]
170	end
171
172	defp server_opts(mode, opts) do
173	namespace = opts[:namespace] \|\| Config.get(:namespace)	7✔
174	[name: opts[:name], namespace: namespace, redis: opts[:redis], mode: mode]
175	end
176
177	defp get_queues(queue_configs) do
178	Enum.map(queue_configs, fn queue_config ->	127✔
179	case queue_config do	138✔
180	{queue, _concurrency} -> queue	7✔
181	queue -> queue	131✔
182	end
183	end)
184	end
185
186	defp get_concurrency(queue_configs, per_queue_concurrency) do
187	Enum.map(queue_configs, fn queue_config ->	127✔
188	case queue_config do	138✔
189	{queue, concurrency} -> {queue, cast_concurrency(concurrency)}	7✔
190	queue -> {queue, per_queue_concurrency}	131✔
191	end
192	end)
193	end
194
195	def cast_concurrency({module, options}), do: {module, options}	1✔
UNCOV 196	def cast_concurrency(:infinity), do: {Exq.Dequeue.Local, [concurrency: :infinity]}	×
197	def cast_concurrency(:infinite), do: {Exq.Dequeue.Local, [concurrency: :infinity]}	111✔
198	def cast_concurrency(x) when is_integer(x), do: {Exq.Dequeue.Local, [concurrency: x]}	20✔
199
200	def cast_concurrency(x) when is_binary(x) do
201	case x \|> String.trim() \|> String.downcase() do	4✔
202	"infinity" -> {Exq.Dequeue.Local, [concurrency: :infinity]}	1✔
203	"infinite" -> {Exq.Dequeue.Local, [concurrency: :infinity]}	1✔
204	x -> {Exq.Dequeue.Local, [concurrency: Coercion.to_integer(x)]}	2✔
205	end
206	end
207	end

akira / exq / 16bf7540e38e434e6ed4fadc1f40894960dd12db-PR-500

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous