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_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, 60 * @minutes}, 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