defmodule NSQ.Config do
@ms 1
@seconds 1000 * @ms
@minutes 60 * @seconds
@default_config %{
dial_timeout: 1000,
# Deadlines for network reads and writes
read_timeout: 1 * @minutes,
write_timeout: 1 * @seconds,
# {host, port} tuples identifying where we should look for nsqd/nsqlookupd.
nsqds: [],
nsqlookupds: [],
# Duration between polling lookupd for new producers, and fractional jitter
# to add to the lookupd pool loop. this helps evenly distribute requests
# even if multiple consumers restart at the same time
#
# NOTE: when not using nsqlookupd, lookupd_poll_interval represents the
# duration of time between reconnection attempts
lookupd_poll_interval: 1 * @minutes,
lookupd_poll_jitter: 0.3,
# If nsqlookupd is not being used and a connection to nsqd should fail,
# it will automatically attempt to reconnect based on the
# lookupd_poll_interval. This is how many times it will make the attempt
# before erroring out.
max_reconnect_attempts: 30,
# Maximum duration when REQueueing (for doubling of deferred requeue)
max_requeue_delay: 15 * @minutes,
# Backoff strategy, defaults to exponential backoff. Overwrite this to
# define alternative backoff algorithms
backoff_strategy: :exponential,
# Maximum amount of time to backoff when processing fails 0 == no backoff
max_backoff_duration: 2 * @minutes,
# Unit of time for calculating consumer backoff
backoff_multiplier: 1 * @seconds,
# Maximum number of times this consumer will attempt to process a message
# before giving up
max_attempts: 5,
# Duration to wait for a message from a producer when in a state where RDY
# counts are re-distributed (ie. max_in_flight < num_producers)
low_rdy_idle_timeout: 10 * @seconds,
# Duration between redistributing max-in-flight to connections
rdy_redistribute_interval: 5 * @seconds,
# Duration to wait to retry RDY, if conn is maxed out
rdy_retry_delay: 5 * @seconds,
# Identifiers sent to nsqd representing this client UserAgent is in the
# spirit of HTTP (default: "<client_library_name>/<version>")
client_id: nil,
hostname: nil,
user_agent: nil,
# Duration of time between heartbeats. This must be less than read_timeout
heartbeat_interval: 30 * @seconds,
# Integer percentage 0-99 to sample the channel (requires nsqd 0.2.25+)
sample_rate: 0,
# To set TLS config, use the following options:
#
# tls_v1 - Bool enable TLS negotiation
# tls_root_ca_file - String path to file containing root CA
# tls_insecure_skip_verify - Bool indicates whether this client should verify server certificates
# tls_cert - String path to file containing public key for certificate
# tls_key - String path to file containing private key for certificate
# tls_min_version - String indicating the minimum version of tls acceptable :sslv3, :tlsv1, :"tlsv1.1", :"tlsv1.2"
tls_v1: false,
tls_insecure_skip_verify: false,
tls_cert: nil,
tls_key: nil,
tls_min_version: nil,
# Compression settings
deflate: false,
deflate_level: 6,
snappy: false,
# Size of the buffer (in bytes) used by nsqd for buffering writes to this
# connection
output_buffer_size: 16384,
# Timeout used by nsqd before flushing buffered writes (set to 0 to disable).
#
# WARNING: configuring clients with an extremely low
# (< 25ms) output_buffer_timeout has a significant effect
# on nsqd CPU usage (particularly with > 50 clients connected).
output_buffer_timeout: 250 * @ms,
# Maximum number of messages to allow in flight (concurrency knob)
max_in_flight: 2500,
# The server-side message timeout for messages delivered to this client.
# After waiting this long without a TOUCH, NSQD will automatically requeue
# the message.
msg_timeout: 60 * @seconds,
# secret for nsqd authentication (requires nsqd 0.2.29+)
auth_secret: "",
# function or module to deal with messages
message_handler: nil,
# define a custom event manager to get callbacks as to what NSQ is doing
event_manager: nil
}
@valid_ranges %{
read_timeout: {100 * @ms, 5 * @minutes},
write_timeout: {100 * @ms, 5 * @minutes},
lookupd_poll_interval: {10 * @ms, 5 * @minutes},
lookupd_poll_jitter: {0, 1},
max_requeue_delay: {0, :infinity},
max_backoff_duration: {0, 60 * @minutes},
backoff_multiplier: {0, 60 * @minutes},
max_attempts: {0, 65535},
low_rdy_idle_timeout: {1 * @seconds, 5 * @minutes},
rdy_redistribute_interval: {1 * @ms, 5 * @seconds},
sample_rate: {0, 99},
deflate_level: {1, 9},
max_in_flight: {0, :infinity},
msg_timeout: {0, :infinity}
}
defstruct Enum.into(@default_config, [])
@doc """
Given a config, tell us what's wrong with it. If nothing is wrong, we'll
return `{:ok, config}`.
## Examples
iex> NSQ.Config.validate(%NSQ.Config{})
{:ok, %NSQ.Config{}}
iex> NSQ.Config.validate(%NSQ.Config{max_attempts: -1})
{:error, ["max_attempts: -1 below minimum 0"]}
"""
def validate(config) do
errors = []
%NSQ.Config{} = config
errors =
errors ++
Enum.map(@valid_ranges, fn {name, {min, max}} ->
case range_error(Map.get(config, name), min, max) do
{:error, reason} -> "#{name}: #{reason}"
:ok -> nil
end
end)
errors = [
no_match_error(
config.backoff_strategy,
[:exponential, :test]
)
| errors
]
errors = Enum.reject(errors, fn v -> v == nil end)
if length(errors) > 0 do
{:error, errors}
else
{:ok, config}
end
end
def normalize(config) do
config = %NSQ.Config{config | nsqds: normalize_hosts(config.nsqds)}
config = %NSQ.Config{config | nsqlookupds: normalize_hosts(config.nsqlookupds)}
{:ok, config}
end
def normalize_hosts(hosts) do
Enum.map(hosts, fn host_with_port ->
cond do
is_tuple(host_with_port) ->
{_host, _port} = host_with_port
is_binary(host_with_port) ->
[host, port] = host_with_port |> String.split(":")
{port, _} = Integer.parse(port)
{host, port}
is_list(host_with_port) ->
{_host, _port} = List.to_tuple(host_with_port)
true ->
raise "Invalid host definition #{inspect(host_with_port)}"
end
end)
end
defp range_error(val, min, max) do
cond do
val == nil -> :ok
val < min -> {:error, "#{val} below minimum #{min}"}
max != :infinity && val > max -> {:error, "#{val} above maximum #{max}"}
true -> :ok
end
end
defp matches_any?(val, candidates) do
Enum.any?(candidates, fn candidate -> candidate == val end)
end
defp no_match_error(val, candidates) do
if matches_any?(val, candidates) do
nil
else
{:error, "#{val} doesn't match any supported values"}
end
end
end
