emqx4.3/priv/emq.schema

%%-*- mode: erlang -*-
%% EMQ R2.3 config mapping

%%--------------------------------------------------------------------
%% Cluster
%%--------------------------------------------------------------------

%% @doc Cluster name
{mapping, "cluster.name", "ekka.cluster_name", [
  {default, emqcl},
  {datatype, atom}
]}.

%% @doc Cluster discovery
{mapping, "cluster.discovery", "ekka.cluster_discovery", [
  {default, manual},
  {datatype, atom}
]}.

%% @doc Clean down node from the cluster
{mapping, "cluster.autoclean", "ekka.cluster_autoclean", [
  {datatype, {duration, ms}}
]}.

%% @doc Cluster autoheal
{mapping, "cluster.autoheal", "ekka.cluster_autoheal", [
  {datatype, flag},
  {default, off}
]}.

%%--------------------------------------------------------------------
%% Cluster by static node list

{mapping, "cluster.static.seeds", "ekka.cluster_discovery", [
  {datatype, string}
]}.

%%--------------------------------------------------------------------
%% Cluster by UDP Multicast

{mapping, "cluster.mcast.addr", "ekka.cluster_discovery", [
  {default, "239.192.0.1"},
  {datatype, string}
]}.

{mapping, "cluster.mcast.ports", "ekka.cluster_discovery", [
  {default, "4369"},
  {datatype, string}
]}.

{mapping, "cluster.mcast.iface", "ekka.cluster_discovery", [
  {datatype, string},
  {default, "0.0.0.0"}
]}.

{mapping, "cluster.mcast.ttl", "ekka.cluster_discovery", [
  {datatype, integer},
  {default, 255}
]}.

{mapping, "cluster.mcast.loop", "ekka.cluster_discovery", [
  {datatype, flag},
  {default, on}
]}.

{mapping, "cluster.mcast.sndbuf", "ekka.cluster_discovery", [
  {datatype, bytesize},
  {default, "16KB"}
]}.

{mapping, "cluster.mcast.recbuf", "ekka.cluster_discovery", [
  {datatype, bytesize},
  {default, "16KB"}
]}.

{mapping, "cluster.mcast.buffer", "ekka.cluster_discovery", [
  {datatype, bytesize},
  {default, "32KB"}
]}.

%%--------------------------------------------------------------------
%% Cluster by DNS A Record

{mapping, "cluster.dns.name", "ekka.cluster_discovery", [
  {datatype, string}
]}.

{mapping, "cluster.dns.app", "ekka.cluster_discovery", [
  {datatype, string}
]}.

%%--------------------------------------------------------------------
%% Cluster using etcd

{mapping, "cluster.etcd.server", "ekka.cluster_discovery", [
  {datatype, string}
]}.

{mapping, "cluster.etcd.prefix", "ekka.cluster_discovery", [
  {datatype, string}
]}.

{mapping, "cluster.etcd.node_ttl", "ekka.cluster_discovery", [
  {datatype, {duration, ms}},
  {default, "1m"}
]}.

%%--------------------------------------------------------------------
%% Cluster on K8s

{mapping, "cluster.k8s.apiserver", "ekka.cluster_discovery", [
  {datatype, string}
]}.

{mapping, "cluster.k8s.service_name", "ekka.cluster_discovery", [
  {datatype, string}
]}.

{mapping, "cluster.k8s.address_type", "ekka.cluster_discovery", [
  {datatype, {enum, [ip, dns]}}
]}.

{mapping, "cluster.k8s.app_name", "ekka.cluster_discovery", [
  {datatype, string}
]}.

{translation, "ekka.cluster_discovery", fun(Conf) ->
  Strategy = cuttlefish:conf_get("cluster.discovery", Conf),
  Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
  IpPort = fun(S) ->
             [Addr, Port] = string:tokens(S, ":"),
             {ok, Ip} = inet:parse_address(Addr),
             {Ip, Port}
           end,
  Options = fun(static) ->
                 [{seeds, [list_to_atom(S) || S <- string:tokens(cuttlefish:conf_get("cluster.static.seeds", Conf, ""), ",")]}];
               (mcast) ->
                 {ok, Addr} = inet:parse_address(cuttlefish:conf_get("cluster.mcast.addr", Conf)),
                 {ok, Iface} = inet:parse_address(cuttlefish:conf_get("cluster.mcast.iface", Conf)),
                 Ports = [list_to_integer(S) || S <- string:tokens(cuttlefish:conf_get("cluster.mcast.ports", Conf), ",")],
                 [{addr, Addr}, {ports, Ports}, {iface, Iface},
                  {ttl, cuttlefish:conf_get("cluster.mcast.ttl", Conf, 1)},
                  {loop, cuttlefish:conf_get("cluster.mcast.loop", Conf, true)}];
               (dns) ->
                 [{name, cuttlefish:conf_get("cluster.dns.name", Conf)},
                  {app, cuttlefish:conf_get("cluster.dns.app", Conf)}];
               (etcd) ->
                 [{server, string:tokens(cuttlefish:conf_get("cluster.etcd.server", Conf), ",")},
                  {prefix, cuttlefish:conf_get("cluster.etcd.prefix", Conf, "emqcl")},
                  {node_ttl, cuttlefish:conf_get("cluster.etcd.node_ttl", Conf, 60)}];
               (k8s) ->
                 [{apiserver, cuttlefish:conf_get("cluster.k8s.apiserver", Conf)},
                  {service_name, cuttlefish:conf_get("cluster.k8s.service_name", Conf)},
                  {address_type, cuttlefish:conf_get("cluster.k8s.address_type", Conf, ip)},
                  {app_name, cuttlefish:conf_get("cluster.k8s.app_name", Conf)}];
               (manual) ->
                 [ ]
            end,
  {Strategy, Filter(Options(Strategy))}
end}.

%%--------------------------------------------------------------------
%% Erlang Node
%%--------------------------------------------------------------------

%% @doc Erlang node name
{mapping, "node.name", "vm_args.-name", [
  {default, "emq@127.0.0.1"}
]}.

%% @doc Secret cookie for distributed erlang node
{mapping, "node.cookie", "vm_args.-setcookie", [
  {default, "emqsecretcookie"}
]}.

%% @doc SMP Support
{mapping, "node.smp", "vm_args.-smp", [
  {default, auto},
  {datatype, {enum, [enable, auto, disable]}},
  hidden
]}.

%% @doc http://erlang.org/doc/man/heart.html
{mapping, "node.heartbeat", "vm_args.-heart", [
  {datatype, flag},
  hidden
]}.

{translation, "vm_args.-heart", fun(Conf) ->
    case cuttlefish:conf_get("node.heartbeat", Conf) of
        true  -> "";
        false -> cuttlefish:invalid("should be 'on' or comment the line!")
    end
end}.

%% @doc Enable Kernel Poll
{mapping, "node.kernel_poll", "vm_args.+K", [
  {default, on},
  {datatype, flag},
  hidden
]}.

%% @doc More information at: http://erlang.org/doc/man/erl.html
{mapping, "node.async_threads", "vm_args.+A", [
  {default, 64},
  {datatype, integer},
  {validators, ["range:0-1024"]}
]}.

%% @doc Erlang Process Limit
{mapping, "node.process_limit", "vm_args.+P", [
  {datatype, integer},
  {default, 256000},
  hidden
]}.

%% Note: OTP R15 and earlier uses -env ERL_MAX_PORTS, R16+ uses +Q
%% @doc The number of concurrent ports/sockets
%% Valid range is 1024-134217727
{mapping, "node.max_ports",
  cuttlefish:otp("R16", "vm_args.+Q", "vm_args.-env ERL_MAX_PORTS"), [
  {default, 262144},
  {datatype, integer},
  {validators, ["range4ports"]}
]}.

{validator, "range4ports", "must be 1024 to 134217727",
 fun(X) -> X >= 1024 andalso X =< 134217727 end}.

%% @doc http://www.erlang.org/doc/man/erl.html#%2bzdbbl
{mapping, "node.dist_buffer_size", "vm_args.+zdbbl", [
  {datatype, bytesize},
  {commented, "32MB"},
  hidden,
  {validators, ["zdbbl_range"]}
]}.

{translation, "vm_args.+zdbbl",
 fun(Conf) ->
  ZDBBL = cuttlefish:conf_get("node.dist_buffer_size", Conf, undefined),
  case ZDBBL of
    undefined -> undefined;
    X when is_integer(X) -> cuttlefish_util:ceiling(X / 1024); %% Bytes to Kilobytes;
    _ -> undefined
  end
 end
}.

{validator, "zdbbl_range", "must be between 1KB and 2097151KB",
 fun(ZDBBL) ->
  %% 2097151KB = 2147482624
  ZDBBL >= 1024 andalso ZDBBL =< 2147482624
 end
}.

%% @doc http://www.erlang.org/doc/man/erlang.html#system_flag-2
{mapping, "node.fullsweep_after", "vm_args.-env ERL_FULLSWEEP_AFTER", [
  {default, 1000},
  {datatype, integer},
  hidden,
  {validators, ["positive_integer"]}
]}.

{validator, "positive_integer", "must be a positive integer",
  fun(X) -> X >= 0 end}.

%% Note: OTP R15 and earlier uses -env ERL_MAX_ETS_TABLES,
%% R16+ uses +e
%% @doc The ETS table limit
{mapping, "node.max_ets_tables",
  cuttlefish:otp("R16", "vm_args.+e", "vm_args.-env ERL_MAX_ETS_TABLES"), [
  {default, 256000},
  {datatype, integer},
  hidden
]}.

%% @doc Set the location of crash dumps
{mapping, "node.crash_dump", "vm_args.-env ERL_CRASH_DUMP", [
  {default, "{{crash_dump}}"},
  {datatype, file},
  hidden
]}.

%% @doc http://www.erlang.org/doc/man/kernel_app.html#net_ticktime
{mapping, "node.dist_net_ticktime", "vm_args.-kernel net_ticktime", [
  {commented, 60},
  {datatype, integer},
  hidden
]}.

%% @doc http://www.erlang.org/doc/man/kernel_app.html
{mapping, "node.dist_listen_min", "kernel.inet_dist_listen_min", [
  {commented, 6369},
  {datatype, integer},
  hidden
]}.

%% @see node.dist_listen_min
{mapping, "node.dist_listen_max", "kernel.inet_dist_listen_max", [
  {commented, 6369},
  {datatype, integer},
  hidden
]}.

%%--------------------------------------------------------------------
%% Log
%%--------------------------------------------------------------------

{mapping, "log.dir", "lager.log_dir", [
  {default, "log"},
  {datatype, string}
]}.

{mapping, "log.console", "lager.handlers", [
  {default, file},
  {datatype, {enum, [off, file, console, both]}}
]}.

{mapping, "log.console.level", "lager.handlers", [
  {default, info},
  {datatype, {enum, [debug, info, notice, warning, error, critical, alert, emergency, none]}}
]}.

{mapping, "log.console.file", "lager.handlers", [
  {default, "log/console.log"},
  {datatype, file}
]}.

{mapping, "log.console.size", "lager.handlers", [
  {default, 10485760},
  {datatype, integer}
]}.

{mapping, "log.console.count", "lager.handlers", [
  {default, 5},
  {datatype, integer}
]}.

{mapping, "log.info.file", "lager.handlers", [
  {datatype, file}
]}.

{mapping, "log.info.size", "lager.handlers", [
  {default, 10485760},
  {datatype, integer}
]}.

{mapping, "log.info.count", "lager.handlers", [
  {default, 5},
  {datatype, integer}
]}.

{mapping, "log.error.file", "lager.handlers", [
  {default, "log/error.log"},
  {datatype, file}
]}.

{mapping, "log.error.size", "lager.handlers", [
  {default, 10485760},
  {datatype, integer}
]}.

{mapping, "log.error.count", "lager.handlers", [
  {default, 5},
  {datatype, integer}
]}.

{mapping, "log.syslog", "lager.handlers", [
  {default,  off},
  {datatype, flag}
]}.

{mapping, "log.syslog.identity", "lager.handlers", [
  {default, "emqttd"},
  {datatype, string}
]}.

{mapping, "log.syslog.facility", "lager.handlers", [
  {default, local0},
  {datatype, {enum, [daemon, local0, local1, local2, local3, local4, local5, local6, local7]}}
]}.

{mapping, "log.syslog.level", "lager.handlers", [
  {default, error},
  {datatype, {enum, [debug, info, notice, warning, error, critical, alert, emergency]}}
]}.

{mapping, "log.error.redirect", "lager.error_logger_redirect", [
  {default, on},
  {datatype, flag},
  hidden
]}.

{mapping, "log.error.messages_per_second", "lager.error_logger_hwm", [
  {default, 1000},
  {datatype, integer},
  hidden
]}.

{translation,
 "lager.handlers",
 fun(Conf) ->
    ErrorHandler = case cuttlefish:conf_get("log.error.file", Conf, undefined) of
      undefined -> [];
      ErrorFilename -> [{lager_file_backend, [{file, ErrorFilename},
                                              {level, error},
                                              {size, cuttlefish:conf_get("log.error.size", Conf)},
                                              {date, "$D0"},
                                              {count, cuttlefish:conf_get("log.error.count", Conf)}]}]
    end,

    InfoHandler = case cuttlefish:conf_get("log.info.file", Conf, undefined) of
      undefined -> [];
      InfoFilename -> [{lager_file_backend, [{file, InfoFilename},
                                             {level, info},
                                             {size, cuttlefish:conf_get("log.info.size", Conf)},
                                             {date, "$D0"},
                                             {count, cuttlefish:conf_get("log.info.count", Conf)}]}]
    end,

    ConsoleLogLevel = cuttlefish:conf_get("log.console.level", Conf),
    ConsoleLogFile = cuttlefish:conf_get("log.console.file", Conf),

    ConsoleHandler = {lager_console_backend, ConsoleLogLevel},
    ConsoleFileHandler = {lager_file_backend, [{file, ConsoleLogFile},
                                               {level, ConsoleLogLevel},
                                               {size, cuttlefish:conf_get("log.console.size", Conf)},
                                               {date, "$D0"},
                                               {count, cuttlefish:conf_get("log.console.count", Conf)}]},

    ConsoleHandlers = case cuttlefish:conf_get("log.console", Conf) of
      off -> [];
      file -> [ConsoleFileHandler];
      console -> [ConsoleHandler];
      both -> [ConsoleHandler, ConsoleFileHandler];
      _ -> []
    end,
    SyslogHandler = case cuttlefish:conf_get("log.syslog", Conf) of
      false -> [];
      true  -> [{lager_syslog_backend,
                  [cuttlefish:conf_get("log.syslog.identity", Conf),
                   cuttlefish:conf_get("log.syslog.facility", Conf),
                   cuttlefish:conf_get("log.syslog.level", Conf)]}]
    end,
    ConsoleHandlers ++ ErrorHandler ++ InfoHandler ++ SyslogHandler
  end
}.

{mapping, "log.crash", "lager.crash_log", [
  {default, on},
  {datatype, flag}
]}.

{mapping, "log.crash.file", "lager.crash_log", [
  {default, "log/crash.log"},
  {datatype, file}
]}.

{translation,
 "lager.crash_log",
 fun(Conf) ->
     case cuttlefish:conf_get("log.crash", Conf) of
         false -> undefined;
         _ ->
             cuttlefish:conf_get("log.crash.file", Conf, "./log/crash.log")
     end
 end}.

{mapping, "sasl", "sasl.sasl_error_logger", [
  {default, off},
  {datatype, flag},
  hidden
]}.

%%--------------------------------------------------------------------
%% Allow Anonymous and Default ACL
%%--------------------------------------------------------------------

%% @doc Allow Anonymous
{mapping, "mqtt.allow_anonymous", "emqttd.allow_anonymous", [
  {default, false},
  {datatype, {enum, [true, false]}}
]}.

%% @doc ACL nomatch
{mapping, "mqtt.acl_nomatch", "emqttd.acl_nomatch", [
  {default, allow},
  {datatype, {enum, [allow, deny]}}
]}.

%% @doc Default ACL File
{mapping, "mqtt.acl_file", "emqttd.acl_file", [
  {datatype, string},
  hidden
]}.

%% @doc Cache ACL for PUBLISH
{mapping, "mqtt.cache_acl", "emqttd.cache_acl", [
  {default, true},
  {datatype, {enum, [true, false]}}
]}.

%%--------------------------------------------------------------------
%% MQTT Protocol
%%--------------------------------------------------------------------

%% @doc Set the Max ClientId Length Allowed.
{mapping, "mqtt.max_clientid_len", "emqttd.protocol", [
  {default, 1024},
  {datatype, integer}
]}.

%% @doc Max Packet Size Allowed, 64K by default.
{mapping, "mqtt.max_packet_size", "emqttd.protocol", [
  {default, "64KB"},
  {datatype, bytesize}
]}.

%% @doc Keepalive backoff
{mapping, "mqtt.keepalive_backoff", "emqttd.protocol", [
  {default, 1.25},
  {datatype, float}
]}.

{translation, "emqttd.protocol", fun(Conf) ->
  [{max_clientid_len, cuttlefish:conf_get("mqtt.max_clientid_len", Conf)},
   {max_packet_size, cuttlefish:conf_get("mqtt.max_packet_size", Conf)},
   {keepalive_backoff, cuttlefish:conf_get("mqtt.keepalive_backoff", Conf)}]
end}.

{mapping, "mqtt.websocket_protocol_header", "emqttd.websocket_protocol_header", [
  {default, on},
  {datatype, flag}
]}.

%%--------------------------------------------------------------------
%% MQTT Connection
%%--------------------------------------------------------------------

%% @doc Force the client to GC: integer
{mapping, "mqtt.conn.force_gc_count", "emqttd.conn_force_gc_count", [
  {datatype, integer}
]}.

%%--------------------------------------------------------------------
%% MQTT Client
%%--------------------------------------------------------------------

%% @doc Max Publish Rate of Message
{mapping, "mqtt.client.max_publish_rate", "emqttd.client", [
  {default, 0},
  {datatype, integer}
]}.

%% @doc Client Idle Timeout.
{mapping, "mqtt.client.idle_timeout", "emqttd.client", [
  {default, "30s"},
  {datatype, {duration, ms}}
]}.

%% @doc Enable Stats of Client.
{mapping, "mqtt.client.enable_stats", "emqttd.client", [
  {default, off},
  {datatype, flag}
]}.

{translation, "emqttd.client", fun(Conf) ->
  [{max_publish_rate, cuttlefish:conf_get("mqtt.client.max_publish_rate", Conf)},
   {client_idle_timeout, cuttlefish:conf_get("mqtt.client.idle_timeout", Conf)},
   {client_enable_stats, cuttlefish:conf_get("mqtt.client.enable_stats", Conf)}]
end}.

%%--------------------------------------------------------------------
%% MQTT Session
%%--------------------------------------------------------------------

%% @doc Max Number of Subscriptions Allowed
{mapping, "mqtt.session.max_subscriptions", "emqttd.session", [
  {default, 0},
  {datatype, integer}
]}.

%% @doc Upgrade QoS?
{mapping, "mqtt.session.upgrade_qos", "emqttd.session", [
  {default, off},
  {datatype, flag}
]}.

%% @doc Max number of QoS 1 and 2 messages that can be “inflight” at one time.
%% 0 means no limit
{mapping, "mqtt.session.max_inflight", "emqttd.session", [
  {default, 100},
  {datatype, integer}
]}.

%% @doc Retry interval for redelivering QoS1/2 messages.
{mapping, "mqtt.session.retry_interval", "emqttd.session", [
  {default, "20s"},
  {datatype, {duration, ms}}
]}.

%% @doc Max Packets that Awaiting PUBREL, 0 means no limit
{mapping, "mqtt.session.max_awaiting_rel", "emqttd.session", [
  {default, 0},
  {datatype, integer}
]}.

%% @doc Awaiting PUBREL Timeout
{mapping, "mqtt.session.await_rel_timeout", "emqttd.session", [
  {default, "20s"},
  {datatype, {duration, ms}}
]}.

%% @doc Enable Stats
{mapping, "mqtt.session.enable_stats", "emqttd.session", [
  {default, off},
  {datatype, flag}
]}.

%% @doc Session Expiry Interval
{mapping, "mqtt.session.expiry_interval", "emqttd.session", [
  {default, "2h"},
  {datatype, {duration, ms}}
]}.

%% @doc Ignore message from self publish
{mapping, "mqtt.session.ignore_loop_deliver", "emqttd.session", [
  {default, false},
  {datatype, {enum, [true, false]}}
]}.

{translation, "emqttd.session", fun(Conf) ->
  [{max_subscriptions, cuttlefish:conf_get("mqtt.session.max_subscriptions", Conf)},
   {upgrade_qos,       cuttlefish:conf_get("mqtt.session.upgrade_qos", Conf)},
   {max_inflight,      cuttlefish:conf_get("mqtt.session.max_inflight", Conf)},
   {retry_interval,    cuttlefish:conf_get("mqtt.session.retry_interval", Conf)},
   {max_awaiting_rel,  cuttlefish:conf_get("mqtt.session.max_awaiting_rel", Conf)},
   {await_rel_timeout, cuttlefish:conf_get("mqtt.session.await_rel_timeout", Conf)},
   {enable_stats,      cuttlefish:conf_get("mqtt.session.enable_stats", Conf)},
   {expiry_interval,   cuttlefish:conf_get("mqtt.session.expiry_interval", Conf)},
   {ignore_loop_deliver, cuttlefish:conf_get("mqtt.session.ignore_loop_deliver", Conf)}]
end}.

%%--------------------------------------------------------------------
%% MQTT MQueue
%%--------------------------------------------------------------------

%% @doc Type: simple | priority
{mapping, "mqtt.mqueue.type", "emqttd.mqueue", [
  {default, simple},
  {datatype, atom}
]}.

%% @doc Topic Priority: 0~255, Default is 0
{mapping, "mqtt.mqueue.priority", "emqttd.mqueue", [
  {default, ""},
  {datatype, string}
]}.

%% @doc Max queue length. Enqueued messages when persistent client disconnected, or inflight window is full. 0 means no limit.
{mapping, "mqtt.mqueue.max_length", "emqttd.mqueue", [
  {default, 0},
  {datatype, integer}
]}.

%% @doc Low-water mark of queued messages
{mapping, "mqtt.mqueue.low_watermark", "emqttd.mqueue", [
  {default, "20%"},
  {datatype, {percent, float}}
]}.

%% @doc High-water mark of queued messages
{mapping, "mqtt.mqueue.high_watermark", "emqttd.mqueue", [
  {default, "60%"},
  {datatype, {percent, float}}
]}.

%% @doc Queue Qos0 messages?
{mapping, "mqtt.mqueue.store_qos0", "emqttd.mqueue", [
  {default, true},
  {datatype, {enum, [true, false]}}
]}.

{translation, "emqttd.mqueue", fun(Conf) ->
  Opts = [{type, cuttlefish:conf_get("mqtt.mqueue.type", Conf, simple)},
          {max_length, cuttlefish:conf_get("mqtt.mqueue.max_length", Conf)},
          {low_watermark, cuttlefish:conf_get("mqtt.mqueue.low_watermark", Conf)},
          {high_watermark, cuttlefish:conf_get("mqtt.mqueue.high_watermark", Conf)},
          {store_qos0, cuttlefish:conf_get("mqtt.mqueue.store_qos0", Conf)}],
  case cuttlefish:conf_get("mqtt.mqueue.priority", Conf) of
    undefined -> Opts;
    V         -> [{priority,
                   [begin [T, P] = string:tokens(S, "="),
                          {T, list_to_integer(P)}
                    end || S <- string:tokens(V, ",")]} | Opts]
  end
end}.

%%--------------------------------------------------------------------
%% MQTT Broker
%%--------------------------------------------------------------------

{mapping, "mqtt.broker.sys_interval", "emqttd.broker_sys_interval", [
  {datatype, {duration, ms}},
  {default, "1m"}
]}.

%%--------------------------------------------------------------------
%% MQTT PubSub
%%--------------------------------------------------------------------

{mapping, "mqtt.pubsub.pool_size", "emqttd.pubsub", [
  {default, 8},
  {datatype, integer}
]}.

{mapping, "mqtt.pubsub.async", "emqttd.pubsub", [
  {default, true},
  {datatype, {enum, [true, false]}}
]}.

{translation, "emqttd.pubsub", fun(Conf) ->
  [{pool_size, cuttlefish:conf_get("mqtt.pubsub.pool_size", Conf)},
   {async, cuttlefish:conf_get("mqtt.pubsub.async", Conf)}]
end}.

%%--------------------------------------------------------------------
%% MQTT Bridge
%%--------------------------------------------------------------------

{mapping, "mqtt.bridge.max_queue_len", "emqttd.bridge", [
  {default, 10000},
  {datatype, integer}
]}.

{mapping, "mqtt.bridge.ping_down_interval", "emqttd.bridge", [
  {datatype, {duration, ms}},
  {default, "1s"}
]}.

{translation, "emqttd.bridge", fun(Conf) ->
  [{max_queue_len, cuttlefish:conf_get("mqtt.bridge.max_queue_len", Conf)},
   {ping_down_interval, cuttlefish:conf_get("mqtt.bridge.ping_down_interval", Conf)}]
end}.

%%-------------------------------------------------------------------
%% MQTT Plugins
%%-------------------------------------------------------------------

{mapping, "mqtt.plugins.etc_dir", "emqttd.plugins_etc_dir", [
  {datatype, string}
]}.

{mapping, "mqtt.plugins.loaded_file", "emqttd.plugins_loaded_file", [
  {datatype, string}
]}.

%%--------------------------------------------------------------------
%% MQTT Listeners
%%--------------------------------------------------------------------

%%--------------------------------------------------------------------
%% TCP Listeners

{mapping, "listener.tcp.$name", "emqttd.listeners", [
  {datatype, [integer, ip]}
]}.

{mapping, "listener.tcp.$name.acceptors", "emqttd.listeners", [
  {default, 8},
  {datatype, integer}
]}.

{mapping, "listener.tcp.$name.max_clients", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.tcp.$name.zone", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.tcp.$name.mountpoint", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.tcp.$name.rate_limit", "emqttd.listeners", [
  {default, undefined},
  {datatype, string}
]}.

{mapping, "listener.tcp.$name.access.$id", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.tcp.$name.proxy_protocol", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.tcp.$name.proxy_protocol_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}}
]}.

{mapping, "listener.tcp.$name.peer_cert_as_username", "emqttd.listeners", [
  {datatype, {enum, [cn, dn]}}
]}.

{mapping, "listener.tcp.$name.backlog", "emqttd.listeners", [
  {datatype, integer},
  {default, 1024}
]}.

{mapping, "listener.tcp.$name.send_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}},
  {default, "15s"}
]}.

{mapping, "listener.tcp.$name.send_timeout_close", "emqttd.listeners", [
  {datatype, flag},
  {default, on}
]}.

{mapping, "listener.tcp.$name.recbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.tcp.$name.sndbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.tcp.$name.buffer", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.tcp.$name.tune_buffer", "emqttd.listeners", [
  {datatype, flag},
  hidden
]}.

{mapping, "listener.tcp.$name.nodelay", "emqttd.listeners", [
  {datatype, {enum, [true, false]}},
  hidden
]}.

%%--------------------------------------------------------------------
%% SSL Listeners

{mapping, "listener.ssl.$name", "emqttd.listeners", [
  {datatype, [integer, ip]}
]}.

{mapping, "listener.ssl.$name.acceptors", "emqttd.listeners", [
  {default, 8},
  {datatype, integer}
]}.

{mapping, "listener.ssl.$name.max_clients", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.ssl.$name.zone", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.mountpoint", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.rate_limit", "emqttd.listeners", [
  {default, undefined},
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.access.$id", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.proxy_protocol", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.ssl.$name.proxy_protocol_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}}
]}.

{mapping, "listener.ssl.$name.backlog", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.ssl.$name.send_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}},
  {default, "15s"}
]}.

{mapping, "listener.ssl.$name.send_timeout_close", "emqttd.listeners", [
  {datatype, flag},
  {default, on}
]}.

{mapping, "listener.ssl.$name.recbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.ssl.$name.sndbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.ssl.$name.buffer", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.ssl.$name.tune_buffer", "emqttd.listeners", [
  {datatype, flag},
  hidden
]}.

{mapping, "listener.ssl.$name.nodelay", "emqttd.listeners", [
  {datatype, {enum, [true, false]}},
  hidden
]}.

{mapping, "listener.ssl.$name.tls_versions", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.ciphers", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.handshake_timeout", "emqttd.listeners", [
  {default, "15s"},
  {datatype, {duration, ms}}
]}.

{mapping, "listener.ssl.$name.dhfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.keyfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.certfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.cacertfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ssl.$name.verify", "emqttd.listeners", [
  {datatype, atom}
]}.

{mapping, "listener.ssl.$name.fail_if_no_peer_cert", "emqttd.listeners", [
  {datatype, {enum, [true, false]}}
]}.

{mapping, "listener.ssl.$name.secure_renegotiate", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.ssl.$name.reuse_sessions", "emqttd.listeners", [
  {default, on},
  {datatype, flag}
]}.

{mapping, "listener.ssl.$name.honor_cipher_order", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.ssl.$name.peer_cert_as_username", "emqttd.listeners", [
  {datatype, {enum, [cn, dn]}}
]}.

%%--------------------------------------------------------------------
%% MQTT/WebSocket Listeners

{mapping, "listener.ws.$name", "emqttd.listeners", [
  {datatype, [integer, ip]}
]}.

{mapping, "listener.ws.$name.acceptors", "emqttd.listeners", [
  {default, 8},
  {datatype, integer}
]}.

{mapping, "listener.ws.$name.max_clients", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.ws.$name.rate_limit", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ws.$name.zone", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ws.$name.mountpoint", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ws.$name.access.$id", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.ws.$name.proxy_protocol", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.ws.$name.proxy_protocol_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}}
]}.

{mapping, "listener.ws.$name.backlog", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.ws.$name.send_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}},
  {default, "15s"}
]}.

{mapping, "listener.ws.$name.send_timeout_close", "emqttd.listeners", [
  {datatype, flag},
  {default, on}
]}.

{mapping, "listener.ws.$name.recbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.ws.$name.sndbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.ws.$name.buffer", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.ws.$name.tune_buffer", "emqttd.listeners", [
  {datatype, flag},
  hidden
]}.

{mapping, "listener.ws.$name.nodelay", "emqttd.listeners", [
  {datatype, {enum, [true, false]}},
  hidden
]}.

%%--------------------------------------------------------------------
%% MQTT/WebSocket/SSL Listeners

{mapping, "listener.wss.$name", "emqttd.listeners", [
  {datatype, [integer, ip]}
]}.

{mapping, "listener.wss.$name.acceptors", "emqttd.listeners", [
  {default, 8},
  {datatype, integer}
]}.

{mapping, "listener.wss.$name.max_clients", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.wss.$name.zone", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.mountpoint", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.rate_limit", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.access.$id", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.proxy_protocol", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.wss.$name.proxy_protocol_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}}
]}.

{mapping, "listener.wss.$name.backlog", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.wss.$name.send_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}},
  {default, "15s"}
]}.

{mapping, "listener.wss.$name.send_timeout_close", "emqttd.listeners", [
  {datatype, flag},
  {default, on}
]}.

{mapping, "listener.wss.$name.recbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.wss.$name.sndbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.wss.$name.buffer", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.wss.$name.tune_buffer", "emqttd.listeners", [
  {datatype, flag},
  hidden
]}.

{mapping, "listener.wss.$name.nodelay", "emqttd.listeners", [
  {datatype, {enum, [true, false]}},
  hidden
]}.

{mapping, "listener.wss.$name.tls_versions", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.ciphers", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.handshake_timeout", "emqttd.listeners", [
  {default, "15s"},
  {datatype, {duration, ms}}
]}.

{mapping, "listener.wss.$name.keyfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.certfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.cacertfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.wss.$name.verify", "emqttd.listeners", [
  {datatype, atom}
]}.

{mapping, "listener.wss.$name.fail_if_no_peer_cert", "emqttd.listeners", [
  {datatype, {enum, [true, false]}}
]}.

{mapping, "listener.wss.$name.secure_renegotiate", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.wss.$name.reuse_sessions", "emqttd.listeners", [
  {default, on},
  {datatype, flag}
]}.

{mapping, "listener.wss.$name.honor_cipher_order", "emqttd.listeners", [
  {datatype, flag}
]}.

{mapping, "listener.wss.$name.peer_cert_as_username", "emqttd.listeners", [
  {datatype, {enum, [cn, dn]}}
]}.

{translation, "emqttd.listeners", fun(Conf) ->

    Filter  = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,

    Atom = fun(undefined) -> undefined; (S) -> list_to_atom(S) end,

    Access = fun(S) ->
                 [A, CIDR] = string:tokens(S, " "),
                 {list_to_atom(A), case CIDR of "all" -> all; _ -> CIDR end}
             end,

    AccOpts = fun(Prefix) ->
                  case cuttlefish_variable:filter_by_prefix(Prefix ++ ".access", Conf) of
                      [] -> [];
                      Rules -> [{access, [Access(Rule) || {_, Rule} <- Rules]}]
                  end
              end,

    MountPoint = fun(undefined) -> undefined; (S) -> list_to_binary(S) end,

    ConnOpts = fun(Prefix) ->
                   Filter([{zone, Atom(cuttlefish:conf_get(Prefix ++ ".zone", Conf, undefined))},
                           {rate_limit, cuttlefish:conf_get(Prefix ++ ".rate_limit", Conf, undefined)},
                           {proxy_protocol, cuttlefish:conf_get(Prefix ++ ".proxy_protocol", Conf, undefined)},
                           {proxy_protocol_timeout, cuttlefish:conf_get(Prefix ++ ".proxy_protocol_timeout", Conf, undefined)},
                           {mountpoint, MountPoint(cuttlefish:conf_get(Prefix ++ ".mountpoint", Conf, undefined))},
                           {peer_cert_as_username, cuttlefish:conf_get(Prefix ++ ".peer_cert_as_username", Conf, undefined)}])
               end,

    LisOpts = fun(Prefix) ->
                  Filter([{acceptors,   cuttlefish:conf_get(Prefix ++ ".acceptors", Conf)},
                          {max_clients, cuttlefish:conf_get(Prefix ++ ".max_clients", Conf)},
                          {tune_buffer, cuttlefish:conf_get(Prefix ++ ".tune_buffer", Conf, undefined)} | AccOpts(Prefix)])
              end,
    TcpOpts = fun(Prefix) ->
                   Filter([{backlog, cuttlefish:conf_get(Prefix ++ ".backlog", Conf, undefined)},
                           {send_timeout, cuttlefish:conf_get(Prefix ++ ".send_timeout", Conf, undefined)},
                           {send_timeout_close, cuttlefish:conf_get(Prefix ++ ".send_timeout_close", Conf, undefined)},
                           {recbuf,  cuttlefish:conf_get(Prefix ++ ".recbuf", Conf, undefined)},
                           {sndbuf,  cuttlefish:conf_get(Prefix ++ ".sndbuf", Conf, undefined)},
                           {buffer,  cuttlefish:conf_get(Prefix ++ ".buffer", Conf, undefined)},
                           {nodelay, cuttlefish:conf_get(Prefix ++ ".nodelay", Conf, true)}])
              end,

    SplitFun = fun(undefined) -> undefined; (S) -> string:tokens(S, ",") end,

    SslOpts = fun(Prefix) ->
                 Versions = case SplitFun(cuttlefish:conf_get(Prefix ++ ".tls_versions", Conf, undefined)) of
                                undefined -> undefined;
                                L -> [list_to_atom(V) || V <- L]
                            end,
                  Filter([{versions, Versions},
                          {ciphers, SplitFun(cuttlefish:conf_get(Prefix ++ ".ciphers", Conf, undefined))},
                          {handshake_timeout, cuttlefish:conf_get(Prefix ++ ".handshake_timeout", Conf, undefined)},
                          {dhfile, cuttlefish:conf_get(Prefix ++ ".dhfile", Conf, undefined)},
                          {keyfile,    cuttlefish:conf_get(Prefix ++ ".keyfile", Conf, undefined)},
                          {certfile,   cuttlefish:conf_get(Prefix ++ ".certfile", Conf, undefined)},
                          {cacertfile, cuttlefish:conf_get(Prefix ++ ".cacertfile", Conf, undefined)},
                          {verify,     cuttlefish:conf_get(Prefix ++ ".verify", Conf, undefined)},
                          {fail_if_no_peer_cert, cuttlefish:conf_get(Prefix ++ ".fail_if_no_peer_cert", Conf, undefined)},
                          {secure_renegotiate, cuttlefish:conf_get(Prefix ++ ".secure_renegotiate", Conf, undefined)},
                          {reuse_sessions, cuttlefish:conf_get(Prefix ++ ".reuse_sessions", Conf, undefined)},
                          {honor_cipher_order, cuttlefish:conf_get(Prefix ++ ".honor_cipher_order", Conf, undefined)}])
              end,

    TcpListeners = fun(Type, Name) ->
                       Prefix = string:join(["listener", Type, Name], "."),
                       case cuttlefish:conf_get(Prefix, Conf, undefined) of
                           undefined ->
                               [];
                           ListenOn ->
                               [{Atom(Type), ListenOn, [{connopts, ConnOpts(Prefix)}, {sockopts, TcpOpts(Prefix)} | LisOpts(Prefix)]}]
                       end
                   end,

    SslListeners = fun(Type, Name) ->
                       Prefix = string:join(["listener", Type, Name], "."),
                       case cuttlefish:conf_get(Prefix, Conf, undefined) of
                           undefined ->
                               [];
                           ListenOn ->
                               [{Atom(Type), ListenOn, [{connopts, ConnOpts(Prefix)},
                                                        {sockopts, TcpOpts(Prefix)},
                                                        {sslopts, SslOpts(Prefix)} | LisOpts(Prefix)]}]
                       end
                   end,

    ApiListeners = fun(Type, Name) ->
                       Prefix = string:join(["listener", Type, Name], "."),
                       case cuttlefish:conf_get(Prefix, Conf, undefined) of
                           undefined ->
                               [];
                           ListenOn ->
                               SslOpts1 = case SslOpts(Prefix) of
                                        [] -> [];
                                        SslOpts0 -> [{sslopts, SslOpts0}]
                                      end,
                               [{Atom(Type), ListenOn, [{connopts, ConnOpts(Prefix)},
                                                        {sockopts, TcpOpts(Prefix)}| LisOpts(Prefix)] ++ SslOpts1}]
                       end
                   end,


    lists:flatten([TcpListeners(Type, Name) || {["listener", Type, Name], ListenOn}
                                               <- cuttlefish_variable:filter_by_prefix("listener.tcp", Conf)
                                               ++ cuttlefish_variable:filter_by_prefix("listener.ws", Conf)]
                  ++
                  [SslListeners(Type, Name) || {["listener", Type, Name], ListenOn}
                                               <- cuttlefish_variable:filter_by_prefix("listener.ssl", Conf)
                                               ++ cuttlefish_variable:filter_by_prefix("listener.wss", Conf)]
                  ++
                  [ApiListeners(Type, Name) || {["listener", Type, Name], ListenOn}
                                               <- cuttlefish_variable:filter_by_prefix("listener.api", Conf)])
end}.

%%--------------------------------------------------------------------
%% MQTT REST API Listeners

{mapping, "listener.api.$name", "emqttd.listeners", [
  {datatype, [integer, ip]}
]}.

{mapping, "listener.api.$name.acceptors", "emqttd.listeners", [
  {default, 8},
  {datatype, integer}
]}.

{mapping, "listener.api.$name.max_clients", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.api.$name.rate_limit", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.api.$name.access.$id", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.api.$name.backlog", "emqttd.listeners", [
  {default, 1024},
  {datatype, integer}
]}.

{mapping, "listener.api.$name.send_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}},
  {default, "15s"}
]}.

{mapping, "listener.api.$name.send_timeout_close", "emqttd.listeners", [
  {datatype, flag},
  {default, on}
]}.

{mapping, "listener.api.$name.recbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.api.$name.sndbuf", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.api.$name.buffer", "emqttd.listeners", [
  {datatype, bytesize},
  hidden
]}.

{mapping, "listener.api.$name.tune_buffer", "emqttd.listeners", [
  {datatype, flag},
  hidden
]}.

{mapping, "listener.api.$name.nodelay", "emqttd.listeners", [
  {datatype, {enum, [true, false]}},
  hidden
]}.

{mapping, "listener.api.$name.handshake_timeout", "emqttd.listeners", [
  {datatype, {duration, ms}}
]}.

{mapping, "listener.api.$name.keyfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.api.$name.certfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.api.$name.cacertfile", "emqttd.listeners", [
  {datatype, string}
]}.

{mapping, "listener.api.$name.verify", "emqttd.listeners", [
  {datatype, atom}
]}.

{mapping, "listener.api.$name.fail_if_no_peer_cert", "emqttd.listeners", [
  {datatype, {enum, [true, false]}}
]}.

%%--------------------------------------------------------------------
%% System Monitor
%%--------------------------------------------------------------------

%% @doc Long GC, don't monitor in production mode for:
%% https://github.com/erlang/otp/blob/feb45017da36be78d4c5784d758ede619fa7bfd3/erts/emulator/beam/erl_gc.c#L421
{mapping, "sysmon.long_gc", "emqttd.sysmon", [
  {default, false},
  {datatype, {enum, [true, false]}}
]}.

%% @doc Long Schedule(ms)
{mapping, "sysmon.long_schedule", "emqttd.sysmon", [
  {default, 1000},
  {datatype, integer}
]}.

%% @doc Large Heap
{mapping, "sysmon.large_heap", "emqttd.sysmon", [
  {default, "8MB"},
  {datatype, bytesize}
]}.

%% @doc Monitor Busy Port
{mapping, "sysmon.busy_port", "emqttd.sysmon", [
  {default, false},
  {datatype, {enum, [true, false]}}
]}.

%% @doc Monitor Busy Dist Port
{mapping, "sysmon.busy_dist_port", "emqttd.sysmon", [
  {default, true},
  {datatype, {enum, [true, false]}}
]}.

{translation, "emqttd.sysmon", fun(Conf) ->
    [{long_gc, cuttlefish:conf_get("sysmon.long_gc", Conf)},
     {long_schedule, cuttlefish:conf_get("sysmon.long_schedule", Conf)},
     {large_heap, cuttlefish:conf_get("sysmon.large_heap", Conf)},
     {busy_port, cuttlefish:conf_get("sysmon.busy_port", Conf)},
     {busy_dist_port, cuttlefish:conf_get("sysmon.busy_dist_port", Conf)}]
end}.