%%%----------------------------------------------------------------------
%%% File    : ejabberd_listener.erl
  Author   :   <  >
%%% Purpose : Manage socket listener
  Created : 16 Nov 2002 by   <  >
%%%
%%%
  ejabberd , Copyright ( C ) 2002 - 2023    ProcessOne
%%%
%%% This program is free software; you can redistribute it and/or
  modify it under the terms of the GNU General Public License as
  published by the Free Software Foundation ; either version 2 of the
%%% License, or (at your option) any later version.
%%%
%%% This program is distributed in the hope that it will be useful,
%%% but WITHOUT ANY WARRANTY; without even the implied warranty of
%%% MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
%%% General Public License for more details.
%%%
  You should have received a copy of the GNU General Public License along
  with this program ; if not , write to the Free Software Foundation , Inc. ,
  51 Franklin Street , Fifth Floor , Boston ,     USA .
%%%
%%%----------------------------------------------------------------------

-module(ejabberd_listener).
-behaviour(supervisor).
-author('').
-author('').

-export([start_link/0, init/1, stop/0, start/3, init/3,
	 start_listeners/0, start_listener/3, stop_listeners/0,
	 add_listener/3, delete_listener/2,
	 config_reloaded/0]).
-export([listen_options/0, listen_opt_type/1, validator/0]).
-export([tls_listeners/0]).

-include("logger.hrl").

-type transport() :: tcp | udp.
-type endpoint()  :: {inet:port_number(), inet:ip_address(), transport()}.
-type list_opts() :: [{atom(), term()}].
-type opts() :: #{atom() => term()}.
-type listener() :: {endpoint(), module(), opts()}.
-type sockmod() :: gen_tcp.
-type socket() :: inet:socket().
-type state() :: term().

-export_type([listener/0]).

-callback start(sockmod(), socket(), state()) ->
    {ok, pid()} | {error, any()} | ignore.
-callback start_link(sockmod(), socket(), state()) ->
    {ok, pid()} | {error, any()} | ignore.
-callback accept(pid()) -> any().
-callback listen_opt_type(atom()) -> econf:validator().
-callback listen_options() -> [{atom(), term()} | atom()].
-callback tcp_init(socket(), list_opts()) -> state().
-callback udp_init(socket(), list_opts()) -> state().

-optional_callbacks([listen_opt_type/1, tcp_init/2, udp_init/2]).

-define(TCP_SEND_TIMEOUT, 15000).

start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

init(_) ->
    _ = ets:new(?MODULE, [named_table, public]),
    ejabberd_hooks:add(config_reloaded, ?MODULE, config_reloaded, 50),
    Listeners = ejabberd_option:listen(),
    {ok, {{one_for_one, 10, 1}, listeners_childspec(Listeners)}}.

stop() ->
    ejabberd_hooks:delete(config_reloaded, ?MODULE, config_reloaded, 50),
    stop_listeners(),
    ejabberd_sup:stop_child(?MODULE).

-spec listeners_childspec([listener()]) -> [supervisor:child_spec()].
listeners_childspec(Listeners) ->
    lists:map(
      fun({EndPoint, Module, Opts}) ->
	      ets:insert(?MODULE, {EndPoint, Module, Opts}),
	      {EndPoint,
	       {?MODULE, start, [EndPoint, Module, Opts]},
	       transient, brutal_kill, worker, [?MODULE]}
      end, Listeners).

-spec start_listeners() -> ok.
start_listeners() ->
    Listeners = ejabberd_option:listen(),
    lists:foreach(
      fun(Spec) ->
	      supervisor:start_child(?MODULE, Spec)
      end, listeners_childspec(Listeners)).

-spec start(endpoint(), module(), opts()) -> term().
start(EndPoint, Module, Opts) ->
    proc_lib:start_link(?MODULE, init, [EndPoint, Module, Opts]).

-spec init(endpoint(), module(), opts()) -> ok.
init({_, _, Transport} = EndPoint, Module, AllOpts) ->
    {ModuleOpts, SockOpts} = split_opts(Transport, AllOpts),
    init(EndPoint, Module, ModuleOpts, SockOpts).

-spec init(endpoint(), module(), opts(), [gen_tcp:option()]) -> ok.
init({Port, _, udp} = EndPoint, Module, Opts, SockOpts) ->
    {Port2, ExtraOpts} = case Port of
			     <<"unix:", Path/binary>> ->
				 SO = lists:keydelete(ip, 1, SockOpts),
				 file:delete(Path),
				 {0, [{ip, {local, Path}} | SO]};
			     _ ->
				 {Port, SockOpts}
			 end,
    ExtraOpts2 = lists:keydelete(send_timeout, 1, ExtraOpts),
    case gen_udp:open(Port2, [binary,
			     {active, false},
			     {reuseaddr, true} |
			     ExtraOpts2]) of
	{ok, Socket} ->
	    case inet:sockname(Socket) of
		{ok, {Addr, Port1}} ->
		    proc_lib:init_ack({ok, self()}),
		    case application:ensure_started(ejabberd) of
			ok ->
			    ?INFO_MSG("Start accepting ~ts connections at ~ts for ~p",
				      [format_transport(udp, Opts),
				       format_endpoint({Port1, Addr, udp}), Module]),
			    Opts1 = opts_to_list(Module, Opts),
			    case erlang:function_exported(Module, udp_init, 2) of
				false ->
				    udp_recv(Socket, Module, Opts1);
				true ->
				    State = Module:udp_init(Socket, Opts1),
				    udp_recv(Socket, Module, State)
			    end;
			{error, _} ->
			    ok
		    end;
		{error, Reason} = Err ->
		    report_socket_error(Reason, EndPoint, Module),
		    proc_lib:init_ack(Err)
	    end;
	{error, Reason} = Err ->
	    report_socket_error(Reason, EndPoint, Module),
	    proc_lib:init_ack(Err)
    end;
init({Port, _, tcp} = EndPoint, Module, Opts, SockOpts) ->
    case listen_tcp(Port, SockOpts) of
	{ok, ListenSocket} ->
	    case inet:sockname(ListenSocket) of
		{ok, {Addr, Port1}} ->
		    proc_lib:init_ack({ok, self()}),
		    case application:ensure_started(ejabberd) of
			ok ->
			    Sup = start_module_sup(Module, Opts),
			    Interval = maps:get(accept_interval, Opts),
			    Proxy = maps:get(use_proxy_protocol, Opts),
			    ?INFO_MSG("Start accepting ~ts connections at ~ts for ~p",
				      [format_transport(tcp, Opts),
				       format_endpoint({Port1, Addr, tcp}), Module]),
			    Opts1 = opts_to_list(Module, Opts),
			    case erlang:function_exported(Module, tcp_init, 2) of
				false ->
				    accept(ListenSocket, Module, Opts1, Sup, Interval, Proxy);
				true ->
				    State = Module:tcp_init(ListenSocket, Opts1),
				    accept(ListenSocket, Module, State, Sup, Interval, Proxy)
			    end;
			{error, _} ->
			    ok
		    end;
		{error, Reason} = Err ->
		    report_socket_error(Reason, EndPoint, Module),
		    proc_lib:init_ack(Err)
	    end;
	{error, Reason} = Err ->
	    report_socket_error(Reason, EndPoint, Module),
	    proc_lib:init_ack(Err)
    end.

-spec listen_tcp(inet:port_number(), [gen_tcp:option()]) ->
	        {ok, inet:socket()} | {error, system_limit | inet:posix()}.
listen_tcp(Port, SockOpts) ->
    {Port2, ExtraOpts} = case Port of
			     <<"unix:", Path/binary>> ->
				 SO = lists:keydelete(ip, 1, SockOpts),
				 file:delete(Path),
				 {0, [{ip, {local, Path}} | SO]};
			     _ ->
				 {Port, SockOpts}
			 end,
    Res = gen_tcp:listen(Port2, [binary,
				{packet, 0},
				{active, false},
				{reuseaddr, true},
				{nodelay, true},
				{send_timeout_close, true},
				{keepalive, true} | ExtraOpts]),
    case Res of
	{ok, ListenSocket} ->
	    {ok, ListenSocket};
	{error, _} = Err ->
	    Err
    end.

-spec split_opts(transport(), opts()) -> {opts(), [gen_tcp:option()]}.
split_opts(Transport, Opts) ->
    maps:fold(
      fun(Opt, Val, {ModOpts, SockOpts}) ->
	      case OptVal = {Opt, Val} of
		  {ip, _} ->
		      {ModOpts, [OptVal|SockOpts]};
		  {backlog, _} when Transport == tcp ->
		      {ModOpts, [OptVal|SockOpts]};
		  {backlog, _} ->
		      {ModOpts, SockOpts};
		  _ ->
		      {ModOpts#{Opt => Val}, SockOpts}
	      end
      end, {#{}, []}, Opts).

-spec accept(inet:socket(), module(), state(), atom(),
	     non_neg_integer(), boolean()) -> no_return().
accept(ListenSocket, Module, State, Sup, Interval, Proxy) ->
    Arity = case erlang:function_exported(Module, start, 3) of
		true -> 3;
		false -> 2
	    end,
    accept(ListenSocket, Module, State, Sup, Interval, Proxy, Arity).

-spec accept(inet:socket(), module(), state(), atom(),
	     non_neg_integer(), boolean(), 2|3) -> no_return().
accept(ListenSocket, Module, State, Sup, Interval, Proxy, Arity) ->
    NewInterval = apply_rate_limit(Interval),
    case gen_tcp:accept(ListenSocket) of
	{ok, Socket} when Proxy ->
	    case proxy_protocol:decode(gen_tcp, Socket, 10000) of
		{error, Err} ->
		    ?ERROR_MSG("(~w) Proxy protocol parsing failed: ~ts",
			       [ListenSocket, format_error(Err)]),
		    gen_tcp:close(Socket);
		{undefined, undefined} ->
		    gen_tcp:close(Socket);
		{{Addr, Port}, {PAddr, PPort}} = SP ->
		    %% THIS IS WRONG
		    State2 = [{sock_peer_name, SP} | State],
		    Receiver = case start_connection(Module, Arity, Socket, State2, Sup) of
				   {ok, RecvPid} ->
				       RecvPid;
				   _ ->
				       gen_tcp:close(Socket),
				       none
			       end,
		    ?INFO_MSG("(~p) Accepted proxied connection ~ts -> ~ts",
			      [Receiver,
			       ejabberd_config:may_hide_data(
				 format_endpoint({PPort, PAddr, tcp})),
			       format_endpoint({Port, Addr, tcp})])
	    end,
	    accept(ListenSocket, Module, State, Sup, NewInterval, Proxy, Arity);
	{ok, Socket} ->
	    case {inet:sockname(Socket), inet:peername(Socket)} of
		{{ok, {Addr, Port}}, {ok, {PAddr, PPort}}} ->
		    Receiver = case start_connection(Module, Arity, Socket, State, Sup) of
				   {ok, RecvPid} ->
				       RecvPid;
				   _ ->
				       gen_tcp:close(Socket),
				       none
			       end,
		    ?INFO_MSG("(~p) Accepted connection ~ts -> ~ts",
			      [Receiver,
			       ejabberd_config:may_hide_data(
				 format_endpoint({PPort, PAddr, tcp})),
			       format_endpoint({Port, Addr, tcp})]);
		_ ->
		    gen_tcp:close(Socket)
	    end,
	    accept(ListenSocket, Module, State, Sup, NewInterval, Proxy, Arity);
	{error, Reason} ->
	    ?ERROR_MSG("(~w) Failed TCP accept: ~ts",
		       [ListenSocket, format_error(Reason)]),
	    accept(ListenSocket, Module, State, Sup, NewInterval, Proxy, Arity)
    end.

-spec udp_recv(inet:socket(), module(), state()) -> no_return().
udp_recv(Socket, Module, State) ->
    case gen_udp:recv(Socket, 0) of
	{ok, {Addr, Port, Packet}} ->
	    case catch Module:udp_recv(Socket, Addr, Port, Packet, State) of
		{'EXIT', Reason} ->
		    ?ERROR_MSG("Failed to process UDP packet:~n"
			       "** Source: {~p, ~p}~n"
			       "** Reason: ~p~n** Packet: ~p",
			       [Addr, Port, Reason, Packet]),
		    udp_recv(Socket, Module, State);
		NewState ->
		    udp_recv(Socket, Module, NewState)
	    end;
	{error, Reason} ->
	    ?ERROR_MSG("Unexpected UDP error: ~ts", [format_error(Reason)]),
	    throw({error, Reason})
    end.

-spec start_connection(module(), 2|3, inet:socket(), state(), atom()) ->
		      {ok, pid()} | {error, any()} | ignore.
start_connection(Module, Arity, Socket, State, Sup) ->
    Res = case Sup of
	      undefined when Arity == 3 ->
		  Module:start(gen_tcp, Socket, State);
	      undefined ->
		  Module:start({gen_tcp, Socket}, State);
	      _ when Arity == 3 ->
		  supervisor:start_child(Sup, [gen_tcp, Socket, State]);
	      _ ->
		  supervisor:start_child(Sup, [{gen_tcp, Socket}, State])
	  end,
    case Res of
	{ok, Pid, preowned_socket} ->
	    Module:accept(Pid),
	    {ok, Pid};
	{ok, Pid} ->
	    case gen_tcp:controlling_process(Socket, Pid) of
		ok ->
		    Module:accept(Pid),
		    {ok, Pid};
		Err ->
		    case Sup of
			undefined ->
			    exit(Pid, kill);
			_ ->
			    supervisor:terminate_child(Sup, Pid)
		    end,
		    Err
	    end;
	Err ->
	    Err
    end.

-spec start_listener(endpoint(), module(), opts()) ->
		    {ok, pid()} | {error, any()}.
start_listener(EndPoint, Module, Opts) ->
    %% It is only required to start the supervisor in some cases.
    %% But it doesn't hurt to attempt to start it for any listener.
    %% So, it's normal (and harmless) that in most cases this
    %% call returns: {error, {already_started, pid()}}
    case start_listener_sup(EndPoint, Module, Opts) of
	{ok, _Pid} = R -> R;
	{error, {{'EXIT', {undef, [{M, _F, _A}|_]}}, _} = Error} ->
	    ?ERROR_MSG("Error starting the ejabberd listener: ~p.~n"
		       "It could not be loaded or is not an ejabberd listener.~n"
		       "Error: ~p~n", [Module, Error]),
	    {error, {module_not_available, M}};
	{error, {already_started, Pid}} ->
	    {ok, Pid};
	{error, Error} ->
	    {error, Error}
    end.

-spec start_module_sup(module(), opts()) -> atom().
start_module_sup(Module, Opts) ->
    case maps:get(supervisor, Opts) of
	true ->
	    Proc = list_to_atom(atom_to_list(Module) ++ "_sup"),
	    ChildSpec =	{Proc, {ejabberd_tmp_sup, start_link, [Proc, Module]},
			 permanent,
			 infinity,
			 supervisor,
			 [ejabberd_tmp_sup]},
	    case supervisor:start_child(ejabberd_sup, ChildSpec) of
		{ok, _} -> Proc;
		{error, {already_started, _}} -> Proc;
		_ -> undefined
	    end;
	false ->
	    undefined
    end.

-spec start_listener_sup(endpoint(), module(), opts()) ->
			{ok, pid()} | {error, any()}.
start_listener_sup(EndPoint, Module, Opts) ->
    ChildSpec = {EndPoint,
		 {?MODULE, start, [EndPoint, Module, Opts]},
		 transient,
		 brutal_kill,
		 worker,
		 [?MODULE]},
    supervisor:start_child(?MODULE, ChildSpec).

-spec stop_listeners() -> ok.
stop_listeners() ->
    Ports = ejabberd_option:listen(),
    lists:foreach(
      fun({PortIpNetp, Module, _Opts}) ->
	      delete_listener(PortIpNetp, Module)
      end,
      Ports).

-spec stop_listener(endpoint(), module(), opts()) -> ok | {error, any()}.
stop_listener({_, _, Transport} = EndPoint, Module, Opts) ->
    case supervisor:terminate_child(?MODULE, EndPoint) of
	ok ->
	    ?INFO_MSG("Stop accepting ~ts connections at ~ts for ~p",
		      [format_transport(Transport, Opts),
		       format_endpoint(EndPoint), Module]),
	    ets:delete(?MODULE, EndPoint),
	    supervisor:delete_child(?MODULE, EndPoint);
	Err ->
	    Err
    end.

-spec add_listener(endpoint(), module(), opts()) -> ok | {error, any()}.
add_listener(EndPoint, Module, Opts) ->
    Opts1 = apply_defaults(Module, Opts),
    case start_listener(EndPoint, Module, Opts1) of
	{ok, _Pid} ->
	    ok;
	{error, {already_started, _Pid}} ->
	    {error, {already_started, EndPoint}};
	{error, Error} ->
	    {error, Error}
    end.

-spec delete_listener(endpoint(), module()) -> ok | {error, any()}.
delete_listener(EndPoint, Module) ->
    try ets:lookup_element(?MODULE, EndPoint, 3) of
	Opts -> stop_listener(EndPoint, Module, Opts)
    catch _:badarg ->
	    ok
    end.

-spec tls_listeners() -> [module()].
tls_listeners() ->
    lists:usort(
      lists:filtermap(
	fun({_, Module, #{tls := true}}) -> {true, Module};
	   ({_, Module, #{starttls := true}}) -> {true, Module};
	   (_) -> false
	end, ets:tab2list(?MODULE))).

-spec config_reloaded() -> ok.
config_reloaded() ->
    New = ejabberd_option:listen(),
    Old = ets:tab2list(?MODULE),
    lists:foreach(
      fun({EndPoint, Module, Opts}) ->
	      case lists:keyfind(EndPoint, 1, New) of
		  false ->
		      stop_listener(EndPoint, Module, Opts);
		  _ ->
		      ok
	      end
      end, Old),
    lists:foreach(
      fun({EndPoint, Module, Opts}) ->
	      case lists:keyfind(EndPoint, 1, Old) of
		  {_, Module, Opts} ->
		      ok;
		  {_, OldModule, OldOpts} ->
		      _ = stop_listener(EndPoint, OldModule, OldOpts),
		      case start_listener(EndPoint, Module, Opts) of
			  {ok, _} ->
			      ets:insert(?MODULE, {EndPoint, Module, Opts});
			  _ ->
			      ok
		      end;
		  false ->
		      case start_listener(EndPoint, Module, Opts) of
			  {ok, _} ->
			      ets:insert(?MODULE, {EndPoint, Module, Opts});
			  _ ->
			      ok
		      end
	      end
      end, New).

-spec report_socket_error(inet:posix(), endpoint(), module()) -> ok.
report_socket_error(Reason, EndPoint, Module) ->
    ?ERROR_MSG("Failed to open socket at ~ts for ~ts: ~ts",
	       [format_endpoint(EndPoint), Module, format_error(Reason)]).

-spec format_error(inet:posix() | atom()) -> string().
format_error(Reason) ->
    case inet:format_error(Reason) of
	"unknown POSIX error" ->
	    atom_to_list(Reason);
	ReasonStr ->
	    ReasonStr
    end.

-spec format_endpoint(endpoint()) -> string().
format_endpoint({Port, IP, _Transport}) ->
    case Port of
	Unix when is_binary(Unix) ->
	    <<"unix:", Unix/binary>>;
	_ ->
	    IPStr = case tuple_size(IP) of
			4 -> inet:ntoa(IP);
			8 -> "[" ++ inet:ntoa(IP) ++ "]"
		    end,
	    IPStr ++ ":" ++ integer_to_list(Port)
    end.

-spec format_transport(transport(), opts()) -> string().
format_transport(Transport, Opts) ->
    case maps:get(tls, Opts, false) of
	true when Transport == tcp -> "TLS";
	true when Transport == udp -> "DTLS";
	false when Transport == tcp -> "TCP";
	false when Transport == udp -> "UDP"
    end.

-spec apply_rate_limit(non_neg_integer()) -> non_neg_integer().
apply_rate_limit(Interval) ->
    NewInterval = receive
		      {rate_limit, AcceptInterval} ->
			  AcceptInterval
		  after 0 ->
			  Interval
		  end,
    case NewInterval of
	0  -> ok;
	Ms when is_integer(Ms) ->
	    timer:sleep(Ms);
        {linear, I1, T1, T2, I2} ->
            {MSec, Sec, _USec} = os:timestamp(),
            TS = MSec * 1000000 + Sec,
            I =
                if
                    TS =< T1 -> I1;
                    TS >= T1 + T2 -> I2;
                    true ->
                        round((I2 - I1) * (TS - T1) / T2 + I1)
                end,
            timer:sleep(I)
    end,
    NewInterval.

-spec validator() -> econf:validator().
validator() ->
    econf:and_then(
      econf:list(
	econf:and_then(
	  econf:options(
	    #{module => listen_opt_type(module),
	      transport => listen_opt_type(transport),
	      '_' => econf:any()},
	    [{required, [module]}]),
	  fun(Opts) ->
		  M = proplists:get_value(module, Opts),
		  T = proplists:get_value(transport, Opts, tcp),
		  (validator(M, T))(Opts)
	  end)),
      fun prepare_opts/1).

-spec validator(module(), transport()) -> econf:validator().
validator(M, T) ->
    Options = listen_options() ++ M:listen_options(),
    Required = lists:usort([Opt || Opt <- Options, is_atom(Opt)]),
    Disallowed = if T == udp ->
			 [backlog, use_proxy_protocol, accept_interval];
		    true ->
			 []
		 end,
    Validator = maps:from_list(
		  lists:map(
		    fun(Opt) ->
			    try {Opt, M:listen_opt_type(Opt)}
			    catch _:_ when M /= ?MODULE ->
				    {Opt, listen_opt_type(Opt)}
			    end
		    end, proplists:get_keys(Options))),
    econf:options(
      Validator,
      [{required, Required}, {disallowed, Disallowed},
       {return, map}, unique]).

-spec prepare_opts([opts()]) -> [listener()].
prepare_opts(Listeners) ->
    check_overlapping_listeners(
      lists:map(
	fun(Opts1) ->
		{Opts2, Opts3} = partition(
				   fun({port, _}) -> true;
				      ({transport, _}) -> true;
				      ({module, _}) -> true;
				      (_) -> false
				   end, Opts1),
		Mod = maps:get(module, Opts2),
		Port = maps:get(port, Opts2),
		Transport = maps:get(transport, Opts2, tcp),
		IP = maps:get(ip, Opts3, {0,0,0,0}),
		Opts4 = apply_defaults(Mod, Opts3),
		{{Port, IP, Transport}, Mod, Opts4}
	end, Listeners)).

-spec check_overlapping_listeners([listener()]) -> [listener()].
check_overlapping_listeners(Listeners) ->
    _ = lists:foldl(
	  fun({{Port, IP, Transport} = Key, _, _}, Acc) ->
		  case lists:member(Key, Acc) of
		      true ->
			  econf:fail({listener_dup, {IP, Port}});
		      false ->
			  ZeroIP = case size(IP) of
				       8 -> {0,0,0,0,0,0,0,0};
				       4 -> {0,0,0,0}
				   end,
			  Key1 = {Port, ZeroIP, Transport},
			  case lists:member(Key1, Acc) of
			      true ->
				  econf:fail({listener_conflict,
					      {IP, Port}, {ZeroIP, Port}});
			      false ->
				  [Key|Acc]
			  end
		  end
	  end, [], Listeners),
    Listeners.

-spec apply_defaults(module(), opts()) -> opts().
apply_defaults(Mod, Opts) ->
    lists:foldl(
      fun({Opt, Default}, M) ->
	      case maps:is_key(Opt, M) of
		  true -> M;
		  false -> M#{Opt => Default}
	      end;
	 (_, M) ->
	      M
      end, Opts, Mod:listen_options() ++ listen_options()).

  Convert options to list with removing defaults
-spec opts_to_list(module(), opts()) -> list_opts().
opts_to_list(Mod, Opts) ->
    Defaults = Mod:listen_options() ++ listen_options(),
    maps:fold(
      fun(Opt, Val, Acc) ->
	      case proplists:get_value(Opt, Defaults) of
		  Val -> Acc;
		  _ -> [{Opt, Val}|Acc]
	      end
      end, [], Opts).

-spec partition(fun(({atom(), term()}) -> boolean()), opts()) -> {opts(), opts()}.
partition(Fun, Opts) ->
    maps:fold(
      fun(Opt, Val, {True, False}) ->
	      case Fun({Opt, Val}) of
		  true -> {True#{Opt => Val}, False};
		  false -> {True, False#{Opt => Val}}
	      end
      end, {#{}, #{}}, Opts).

-spec listen_opt_type(atom()) -> econf:validator().
listen_opt_type(port) ->
    econf:either(
	econf:int(0, 65535),
	econf:binary("^unix:.*"));
listen_opt_type(module) ->
    econf:beam([[{start, 3}, {start, 2}],
		[{start_link, 3}, {start_link, 2}],
		{accept, 1}, {listen_options, 0}]);
listen_opt_type(ip) ->
    econf:ip();
listen_opt_type(transport) ->
    econf:enum([tcp, udp]);
listen_opt_type(accept_interval) ->
    econf:non_neg_int();
listen_opt_type(backlog) ->
    econf:non_neg_int();
listen_opt_type(supervisor) ->
    econf:bool();
listen_opt_type(ciphers) ->
    econf:binary();
listen_opt_type(dhfile) ->
    econf:file();
listen_opt_type(cafile) ->
    econf:pem();
listen_opt_type(certfile) ->
    econf:pem();
listen_opt_type(protocol_options) ->
    econf:and_then(
      econf:list(econf:binary()),
      fun(Options) -> str:join(Options, <<"|">>) end);
listen_opt_type(tls_compression) ->
    econf:bool();
listen_opt_type(tls) ->
    econf:bool();
listen_opt_type(max_stanza_size) ->
    econf:pos_int(infinity);
listen_opt_type(max_fsm_queue) ->
    econf:pos_int();
listen_opt_type(send_timeout) ->
    econf:timeout(second, infinity);
listen_opt_type(shaper) ->
    econf:shaper();
listen_opt_type(access) ->
    econf:acl();
listen_opt_type(use_proxy_protocol) ->
    econf:bool().

listen_options() ->
    [module, port,
     {transport, tcp},
     {ip, {0,0,0,0}},
     {accept_interval, 0},
     {send_timeout, 15000},
     {backlog, 5},
     {use_proxy_protocol, false},
     {supervisor, true}].

----------------------------------------------------------------------
 File    : ejabberd_listener.erl
 Purpose : Manage socket listener



 This program is free software; you can redistribute it and/or
 License, or (at your option) any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 General Public License for more details.


----------------------------------------------------------------------
 THIS IS WRONG
 It is only required to start the supervisor in some cases.
 But it doesn't hurt to attempt to start it for any listener.
 So, it's normal (and harmless) that in most cases this
 call returns: {error, {already_started, pid()}}
  Author   :   <  >
  Created : 16 Nov 2002 by   <  >
  ejabberd , Copyright ( C ) 2002 - 2023    ProcessOne
  modify it under the terms of the GNU General Public License as
  published by the Free Software Foundation ; either version 2 of the
  You should have received a copy of the GNU General Public License along
  with this program ; if not , write to the Free Software Foundation , Inc. ,
  51 Franklin Street , Fifth Floor , Boston ,     USA .

-module(ejabberd_listener).
-behaviour(supervisor).
-author('').
-author('').

-export([start_link/0, init/1, stop/0, start/3, init/3,
	 start_listeners/0, start_listener/3, stop_listeners/0,
	 add_listener/3, delete_listener/2,
	 config_reloaded/0]).
-export([listen_options/0, listen_opt_type/1, validator/0]).
-export([tls_listeners/0]).

-include("logger.hrl").

-type transport() :: tcp | udp.
-type endpoint()  :: {inet:port_number(), inet:ip_address(), transport()}.
-type list_opts() :: [{atom(), term()}].
-type opts() :: #{atom() => term()}.
-type listener() :: {endpoint(), module(), opts()}.
-type sockmod() :: gen_tcp.
-type socket() :: inet:socket().
-type state() :: term().

-export_type([listener/0]).

-callback start(sockmod(), socket(), state()) ->
    {ok, pid()} | {error, any()} | ignore.
-callback start_link(sockmod(), socket(), state()) ->
    {ok, pid()} | {error, any()} | ignore.
-callback accept(pid()) -> any().
-callback listen_opt_type(atom()) -> econf:validator().
-callback listen_options() -> [{atom(), term()} | atom()].
-callback tcp_init(socket(), list_opts()) -> state().
-callback udp_init(socket(), list_opts()) -> state().

-optional_callbacks([listen_opt_type/1, tcp_init/2, udp_init/2]).

-define(TCP_SEND_TIMEOUT, 15000).

start_link() ->
    supervisor:start_link({local, ?MODULE}, ?MODULE, []).

init(_) ->
    _ = ets:new(?MODULE, [named_table, public]),
    ejabberd_hooks:add(config_reloaded, ?MODULE, config_reloaded, 50),
    Listeners = ejabberd_option:listen(),
    {ok, {{one_for_one, 10, 1}, listeners_childspec(Listeners)}}.

stop() ->
    ejabberd_hooks:delete(config_reloaded, ?MODULE, config_reloaded, 50),
    stop_listeners(),
    ejabberd_sup:stop_child(?MODULE).

-spec listeners_childspec([listener()]) -> [supervisor:child_spec()].
listeners_childspec(Listeners) ->
    lists:map(
      fun({EndPoint, Module, Opts}) ->
	      ets:insert(?MODULE, {EndPoint, Module, Opts}),
	      {EndPoint,
	       {?MODULE, start, [EndPoint, Module, Opts]},
	       transient, brutal_kill, worker, [?MODULE]}
      end, Listeners).

-spec start_listeners() -> ok.
start_listeners() ->
    Listeners = ejabberd_option:listen(),
    lists:foreach(
      fun(Spec) ->
	      supervisor:start_child(?MODULE, Spec)
      end, listeners_childspec(Listeners)).

-spec start(endpoint(), module(), opts()) -> term().
start(EndPoint, Module, Opts) ->
    proc_lib:start_link(?MODULE, init, [EndPoint, Module, Opts]).

-spec init(endpoint(), module(), opts()) -> ok.
init({_, _, Transport} = EndPoint, Module, AllOpts) ->
    {ModuleOpts, SockOpts} = split_opts(Transport, AllOpts),
    init(EndPoint, Module, ModuleOpts, SockOpts).

-spec init(endpoint(), module(), opts(), [gen_tcp:option()]) -> ok.
init({Port, _, udp} = EndPoint, Module, Opts, SockOpts) ->
    {Port2, ExtraOpts} = case Port of
			     <<"unix:", Path/binary>> ->
				 SO = lists:keydelete(ip, 1, SockOpts),
				 file:delete(Path),
				 {0, [{ip, {local, Path}} | SO]};
			     _ ->
				 {Port, SockOpts}
			 end,
    ExtraOpts2 = lists:keydelete(send_timeout, 1, ExtraOpts),
    case gen_udp:open(Port2, [binary,
			     {active, false},
			     {reuseaddr, true} |
			     ExtraOpts2]) of
	{ok, Socket} ->
	    case inet:sockname(Socket) of
		{ok, {Addr, Port1}} ->
		    proc_lib:init_ack({ok, self()}),
		    case application:ensure_started(ejabberd) of
			ok ->
			    ?INFO_MSG("Start accepting ~ts connections at ~ts for ~p",
				      [format_transport(udp, Opts),
				       format_endpoint({Port1, Addr, udp}), Module]),
			    Opts1 = opts_to_list(Module, Opts),
			    case erlang:function_exported(Module, udp_init, 2) of
				false ->
				    udp_recv(Socket, Module, Opts1);
				true ->
				    State = Module:udp_init(Socket, Opts1),
				    udp_recv(Socket, Module, State)
			    end;
			{error, _} ->
			    ok
		    end;
		{error, Reason} = Err ->
		    report_socket_error(Reason, EndPoint, Module),
		    proc_lib:init_ack(Err)
	    end;
	{error, Reason} = Err ->
	    report_socket_error(Reason, EndPoint, Module),
	    proc_lib:init_ack(Err)
    end;
init({Port, _, tcp} = EndPoint, Module, Opts, SockOpts) ->
    case listen_tcp(Port, SockOpts) of
	{ok, ListenSocket} ->
	    case inet:sockname(ListenSocket) of
		{ok, {Addr, Port1}} ->
		    proc_lib:init_ack({ok, self()}),
		    case application:ensure_started(ejabberd) of
			ok ->
			    Sup = start_module_sup(Module, Opts),
			    Interval = maps:get(accept_interval, Opts),
			    Proxy = maps:get(use_proxy_protocol, Opts),
			    ?INFO_MSG("Start accepting ~ts connections at ~ts for ~p",
				      [format_transport(tcp, Opts),
				       format_endpoint({Port1, Addr, tcp}), Module]),
			    Opts1 = opts_to_list(Module, Opts),
			    case erlang:function_exported(Module, tcp_init, 2) of
				false ->
				    accept(ListenSocket, Module, Opts1, Sup, Interval, Proxy);
				true ->
				    State = Module:tcp_init(ListenSocket, Opts1),
				    accept(ListenSocket, Module, State, Sup, Interval, Proxy)
			    end;
			{error, _} ->
			    ok
		    end;
		{error, Reason} = Err ->
		    report_socket_error(Reason, EndPoint, Module),
		    proc_lib:init_ack(Err)
	    end;
	{error, Reason} = Err ->
	    report_socket_error(Reason, EndPoint, Module),
	    proc_lib:init_ack(Err)
    end.

-spec listen_tcp(inet:port_number(), [gen_tcp:option()]) ->
	        {ok, inet:socket()} | {error, system_limit | inet:posix()}.
listen_tcp(Port, SockOpts) ->
    {Port2, ExtraOpts} = case Port of
			     <<"unix:", Path/binary>> ->
				 SO = lists:keydelete(ip, 1, SockOpts),
				 file:delete(Path),
				 {0, [{ip, {local, Path}} | SO]};
			     _ ->
				 {Port, SockOpts}
			 end,
    Res = gen_tcp:listen(Port2, [binary,
				{packet, 0},
				{active, false},
				{reuseaddr, true},
				{nodelay, true},
				{send_timeout_close, true},
				{keepalive, true} | ExtraOpts]),
    case Res of
	{ok, ListenSocket} ->
	    {ok, ListenSocket};
	{error, _} = Err ->
	    Err
    end.

-spec split_opts(transport(), opts()) -> {opts(), [gen_tcp:option()]}.
split_opts(Transport, Opts) ->
    maps:fold(
      fun(Opt, Val, {ModOpts, SockOpts}) ->
	      case OptVal = {Opt, Val} of
		  {ip, _} ->
		      {ModOpts, [OptVal|SockOpts]};
		  {backlog, _} when Transport == tcp ->
		      {ModOpts, [OptVal|SockOpts]};
		  {backlog, _} ->
		      {ModOpts, SockOpts};
		  _ ->
		      {ModOpts#{Opt => Val}, SockOpts}
	      end
      end, {#{}, []}, Opts).

-spec accept(inet:socket(), module(), state(), atom(),
	     non_neg_integer(), boolean()) -> no_return().
accept(ListenSocket, Module, State, Sup, Interval, Proxy) ->
    Arity = case erlang:function_exported(Module, start, 3) of
		true -> 3;
		false -> 2
	    end,
    accept(ListenSocket, Module, State, Sup, Interval, Proxy, Arity).

-spec accept(inet:socket(), module(), state(), atom(),
	     non_neg_integer(), boolean(), 2|3) -> no_return().
accept(ListenSocket, Module, State, Sup, Interval, Proxy, Arity) ->
    NewInterval = apply_rate_limit(Interval),
    case gen_tcp:accept(ListenSocket) of
	{ok, Socket} when Proxy ->
	    case proxy_protocol:decode(gen_tcp, Socket, 10000) of
		{error, Err} ->
		    ?ERROR_MSG("(~w) Proxy protocol parsing failed: ~ts",
			       [ListenSocket, format_error(Err)]),
		    gen_tcp:close(Socket);
		{undefined, undefined} ->
		    gen_tcp:close(Socket);
		{{Addr, Port}, {PAddr, PPort}} = SP ->
		    State2 = [{sock_peer_name, SP} | State],
		    Receiver = case start_connection(Module, Arity, Socket, State2, Sup) of
				   {ok, RecvPid} ->
				       RecvPid;
				   _ ->
				       gen_tcp:close(Socket),
				       none
			       end,
		    ?INFO_MSG("(~p) Accepted proxied connection ~ts -> ~ts",
			      [Receiver,
			       ejabberd_config:may_hide_data(
				 format_endpoint({PPort, PAddr, tcp})),
			       format_endpoint({Port, Addr, tcp})])
	    end,
	    accept(ListenSocket, Module, State, Sup, NewInterval, Proxy, Arity);
	{ok, Socket} ->
	    case {inet:sockname(Socket), inet:peername(Socket)} of
		{{ok, {Addr, Port}}, {ok, {PAddr, PPort}}} ->
		    Receiver = case start_connection(Module, Arity, Socket, State, Sup) of
				   {ok, RecvPid} ->
				       RecvPid;
				   _ ->
				       gen_tcp:close(Socket),
				       none
			       end,
		    ?INFO_MSG("(~p) Accepted connection ~ts -> ~ts",
			      [Receiver,
			       ejabberd_config:may_hide_data(
				 format_endpoint({PPort, PAddr, tcp})),
			       format_endpoint({Port, Addr, tcp})]);
		_ ->
		    gen_tcp:close(Socket)
	    end,
	    accept(ListenSocket, Module, State, Sup, NewInterval, Proxy, Arity);
	{error, Reason} ->
	    ?ERROR_MSG("(~w) Failed TCP accept: ~ts",
		       [ListenSocket, format_error(Reason)]),
	    accept(ListenSocket, Module, State, Sup, NewInterval, Proxy, Arity)
    end.

-spec udp_recv(inet:socket(), module(), state()) -> no_return().
udp_recv(Socket, Module, State) ->
    case gen_udp:recv(Socket, 0) of
	{ok, {Addr, Port, Packet}} ->
	    case catch Module:udp_recv(Socket, Addr, Port, Packet, State) of
		{'EXIT', Reason} ->
		    ?ERROR_MSG("Failed to process UDP packet:~n"
			       "** Source: {~p, ~p}~n"
			       "** Reason: ~p~n** Packet: ~p",
			       [Addr, Port, Reason, Packet]),
		    udp_recv(Socket, Module, State);
		NewState ->
		    udp_recv(Socket, Module, NewState)
	    end;
	{error, Reason} ->
	    ?ERROR_MSG("Unexpected UDP error: ~ts", [format_error(Reason)]),
	    throw({error, Reason})
    end.

-spec start_connection(module(), 2|3, inet:socket(), state(), atom()) ->
		      {ok, pid()} | {error, any()} | ignore.
start_connection(Module, Arity, Socket, State, Sup) ->
    Res = case Sup of
	      undefined when Arity == 3 ->
		  Module:start(gen_tcp, Socket, State);
	      undefined ->
		  Module:start({gen_tcp, Socket}, State);
	      _ when Arity == 3 ->
		  supervisor:start_child(Sup, [gen_tcp, Socket, State]);
	      _ ->
		  supervisor:start_child(Sup, [{gen_tcp, Socket}, State])
	  end,
    case Res of
	{ok, Pid, preowned_socket} ->
	    Module:accept(Pid),
	    {ok, Pid};
	{ok, Pid} ->
	    case gen_tcp:controlling_process(Socket, Pid) of
		ok ->
		    Module:accept(Pid),
		    {ok, Pid};
		Err ->
		    case Sup of
			undefined ->
			    exit(Pid, kill);
			_ ->
			    supervisor:terminate_child(Sup, Pid)
		    end,
		    Err
	    end;
	Err ->
	    Err
    end.

-spec start_listener(endpoint(), module(), opts()) ->
		    {ok, pid()} | {error, any()}.
start_listener(EndPoint, Module, Opts) ->
    case start_listener_sup(EndPoint, Module, Opts) of
	{ok, _Pid} = R -> R;
	{error, {{'EXIT', {undef, [{M, _F, _A}|_]}}, _} = Error} ->
	    ?ERROR_MSG("Error starting the ejabberd listener: ~p.~n"
		       "It could not be loaded or is not an ejabberd listener.~n"
		       "Error: ~p~n", [Module, Error]),
	    {error, {module_not_available, M}};
	{error, {already_started, Pid}} ->
	    {ok, Pid};
	{error, Error} ->
	    {error, Error}
    end.

-spec start_module_sup(module(), opts()) -> atom().
start_module_sup(Module, Opts) ->
    case maps:get(supervisor, Opts) of
	true ->
	    Proc = list_to_atom(atom_to_list(Module) ++ "_sup"),
	    ChildSpec =	{Proc, {ejabberd_tmp_sup, start_link, [Proc, Module]},
			 permanent,
			 infinity,
			 supervisor,
			 [ejabberd_tmp_sup]},
	    case supervisor:start_child(ejabberd_sup, ChildSpec) of
		{ok, _} -> Proc;
		{error, {already_started, _}} -> Proc;
		_ -> undefined
	    end;
	false ->
	    undefined
    end.

-spec start_listener_sup(endpoint(), module(), opts()) ->
			{ok, pid()} | {error, any()}.
start_listener_sup(EndPoint, Module, Opts) ->
    ChildSpec = {EndPoint,
		 {?MODULE, start, [EndPoint, Module, Opts]},
		 transient,
		 brutal_kill,
		 worker,
		 [?MODULE]},
    supervisor:start_child(?MODULE, ChildSpec).

-spec stop_listeners() -> ok.
stop_listeners() ->
    Ports = ejabberd_option:listen(),
    lists:foreach(
      fun({PortIpNetp, Module, _Opts}) ->
	      delete_listener(PortIpNetp, Module)
      end,
      Ports).

-spec stop_listener(endpoint(), module(), opts()) -> ok | {error, any()}.
stop_listener({_, _, Transport} = EndPoint, Module, Opts) ->
    case supervisor:terminate_child(?MODULE, EndPoint) of
	ok ->
	    ?INFO_MSG("Stop accepting ~ts connections at ~ts for ~p",
		      [format_transport(Transport, Opts),
		       format_endpoint(EndPoint), Module]),
	    ets:delete(?MODULE, EndPoint),
	    supervisor:delete_child(?MODULE, EndPoint);
	Err ->
	    Err
    end.

-spec add_listener(endpoint(), module(), opts()) -> ok | {error, any()}.
add_listener(EndPoint, Module, Opts) ->
    Opts1 = apply_defaults(Module, Opts),
    case start_listener(EndPoint, Module, Opts1) of
	{ok, _Pid} ->
	    ok;
	{error, {already_started, _Pid}} ->
	    {error, {already_started, EndPoint}};
	{error, Error} ->
	    {error, Error}
    end.

-spec delete_listener(endpoint(), module()) -> ok | {error, any()}.
delete_listener(EndPoint, Module) ->
    try ets:lookup_element(?MODULE, EndPoint, 3) of
	Opts -> stop_listener(EndPoint, Module, Opts)
    catch _:badarg ->
	    ok
    end.

-spec tls_listeners() -> [module()].
tls_listeners() ->
    lists:usort(
      lists:filtermap(
	fun({_, Module, #{tls := true}}) -> {true, Module};
	   ({_, Module, #{starttls := true}}) -> {true, Module};
	   (_) -> false
	end, ets:tab2list(?MODULE))).

-spec config_reloaded() -> ok.
config_reloaded() ->
    New = ejabberd_option:listen(),
    Old = ets:tab2list(?MODULE),
    lists:foreach(
      fun({EndPoint, Module, Opts}) ->
	      case lists:keyfind(EndPoint, 1, New) of
		  false ->
		      stop_listener(EndPoint, Module, Opts);
		  _ ->
		      ok
	      end
      end, Old),
    lists:foreach(
      fun({EndPoint, Module, Opts}) ->
	      case lists:keyfind(EndPoint, 1, Old) of
		  {_, Module, Opts} ->
		      ok;
		  {_, OldModule, OldOpts} ->
		      _ = stop_listener(EndPoint, OldModule, OldOpts),
		      case start_listener(EndPoint, Module, Opts) of
			  {ok, _} ->
			      ets:insert(?MODULE, {EndPoint, Module, Opts});
			  _ ->
			      ok
		      end;
		  false ->
		      case start_listener(EndPoint, Module, Opts) of
			  {ok, _} ->
			      ets:insert(?MODULE, {EndPoint, Module, Opts});
			  _ ->
			      ok
		      end
	      end
      end, New).

-spec report_socket_error(inet:posix(), endpoint(), module()) -> ok.
report_socket_error(Reason, EndPoint, Module) ->
    ?ERROR_MSG("Failed to open socket at ~ts for ~ts: ~ts",
	       [format_endpoint(EndPoint), Module, format_error(Reason)]).

-spec format_error(inet:posix() | atom()) -> string().
format_error(Reason) ->
    case inet:format_error(Reason) of
	"unknown POSIX error" ->
	    atom_to_list(Reason);
	ReasonStr ->
	    ReasonStr
    end.

-spec format_endpoint(endpoint()) -> string().
format_endpoint({Port, IP, _Transport}) ->
    case Port of
	Unix when is_binary(Unix) ->
	    <<"unix:", Unix/binary>>;
	_ ->
	    IPStr = case tuple_size(IP) of
			4 -> inet:ntoa(IP);
			8 -> "[" ++ inet:ntoa(IP) ++ "]"
		    end,
	    IPStr ++ ":" ++ integer_to_list(Port)
    end.

-spec format_transport(transport(), opts()) -> string().
format_transport(Transport, Opts) ->
    case maps:get(tls, Opts, false) of
	true when Transport == tcp -> "TLS";
	true when Transport == udp -> "DTLS";
	false when Transport == tcp -> "TCP";
	false when Transport == udp -> "UDP"
    end.

-spec apply_rate_limit(non_neg_integer()) -> non_neg_integer().
apply_rate_limit(Interval) ->
    NewInterval = receive
		      {rate_limit, AcceptInterval} ->
			  AcceptInterval
		  after 0 ->
			  Interval
		  end,
    case NewInterval of
	0  -> ok;
	Ms when is_integer(Ms) ->
	    timer:sleep(Ms);
        {linear, I1, T1, T2, I2} ->
            {MSec, Sec, _USec} = os:timestamp(),
            TS = MSec * 1000000 + Sec,
            I =
                if
                    TS =< T1 -> I1;
                    TS >= T1 + T2 -> I2;
                    true ->
                        round((I2 - I1) * (TS - T1) / T2 + I1)
                end,
            timer:sleep(I)
    end,
    NewInterval.

-spec validator() -> econf:validator().
validator() ->
    econf:and_then(
      econf:list(
	econf:and_then(
	  econf:options(
	    #{module => listen_opt_type(module),
	      transport => listen_opt_type(transport),
	      '_' => econf:any()},
	    [{required, [module]}]),
	  fun(Opts) ->
		  M = proplists:get_value(module, Opts),
		  T = proplists:get_value(transport, Opts, tcp),
		  (validator(M, T))(Opts)
	  end)),
      fun prepare_opts/1).

-spec validator(module(), transport()) -> econf:validator().
validator(M, T) ->
    Options = listen_options() ++ M:listen_options(),
    Required = lists:usort([Opt || Opt <- Options, is_atom(Opt)]),
    Disallowed = if T == udp ->
			 [backlog, use_proxy_protocol, accept_interval];
		    true ->
			 []
		 end,
    Validator = maps:from_list(
		  lists:map(
		    fun(Opt) ->
			    try {Opt, M:listen_opt_type(Opt)}
			    catch _:_ when M /= ?MODULE ->
				    {Opt, listen_opt_type(Opt)}
			    end
		    end, proplists:get_keys(Options))),
    econf:options(
      Validator,
      [{required, Required}, {disallowed, Disallowed},
       {return, map}, unique]).

-spec prepare_opts([opts()]) -> [listener()].
prepare_opts(Listeners) ->
    check_overlapping_listeners(
      lists:map(
	fun(Opts1) ->
		{Opts2, Opts3} = partition(
				   fun({port, _}) -> true;
				      ({transport, _}) -> true;
				      ({module, _}) -> true;
				      (_) -> false
				   end, Opts1),
		Mod = maps:get(module, Opts2),
		Port = maps:get(port, Opts2),
		Transport = maps:get(transport, Opts2, tcp),
		IP = maps:get(ip, Opts3, {0,0,0,0}),
		Opts4 = apply_defaults(Mod, Opts3),
		{{Port, IP, Transport}, Mod, Opts4}
	end, Listeners)).

-spec check_overlapping_listeners([listener()]) -> [listener()].
check_overlapping_listeners(Listeners) ->
    _ = lists:foldl(
	  fun({{Port, IP, Transport} = Key, _, _}, Acc) ->
		  case lists:member(Key, Acc) of
		      true ->
			  econf:fail({listener_dup, {IP, Port}});
		      false ->
			  ZeroIP = case size(IP) of
				       8 -> {0,0,0,0,0,0,0,0};
				       4 -> {0,0,0,0}
				   end,
			  Key1 = {Port, ZeroIP, Transport},
			  case lists:member(Key1, Acc) of
			      true ->
				  econf:fail({listener_conflict,
					      {IP, Port}, {ZeroIP, Port}});
			      false ->
				  [Key|Acc]
			  end
		  end
	  end, [], Listeners),
    Listeners.

-spec apply_defaults(module(), opts()) -> opts().
apply_defaults(Mod, Opts) ->
    lists:foldl(
      fun({Opt, Default}, M) ->
	      case maps:is_key(Opt, M) of
		  true -> M;
		  false -> M#{Opt => Default}
	      end;
	 (_, M) ->
	      M
      end, Opts, Mod:listen_options() ++ listen_options()).

  Convert options to list with removing defaults
-spec opts_to_list(module(), opts()) -> list_opts().
opts_to_list(Mod, Opts) ->
    Defaults = Mod:listen_options() ++ listen_options(),
    maps:fold(
      fun(Opt, Val, Acc) ->
	      case proplists:get_value(Opt, Defaults) of
		  Val -> Acc;
		  _ -> [{Opt, Val}|Acc]
	      end
      end, [], Opts).

-spec partition(fun(({atom(), term()}) -> boolean()), opts()) -> {opts(), opts()}.
partition(Fun, Opts) ->
    maps:fold(
      fun(Opt, Val, {True, False}) ->
	      case Fun({Opt, Val}) of
		  true -> {True#{Opt => Val}, False};
		  false -> {True, False#{Opt => Val}}
	      end
      end, {#{}, #{}}, Opts).

-spec listen_opt_type(atom()) -> econf:validator().
listen_opt_type(port) ->
    econf:either(
	econf:int(0, 65535),
	econf:binary("^unix:.*"));
listen_opt_type(module) ->
    econf:beam([[{start, 3}, {start, 2}],
		[{start_link, 3}, {start_link, 2}],
		{accept, 1}, {listen_options, 0}]);
listen_opt_type(ip) ->
    econf:ip();
listen_opt_type(transport) ->
    econf:enum([tcp, udp]);
listen_opt_type(accept_interval) ->
    econf:non_neg_int();
listen_opt_type(backlog) ->
    econf:non_neg_int();
listen_opt_type(supervisor) ->
    econf:bool();
listen_opt_type(ciphers) ->
    econf:binary();
listen_opt_type(dhfile) ->
    econf:file();
listen_opt_type(cafile) ->
    econf:pem();
listen_opt_type(certfile) ->
    econf:pem();
listen_opt_type(protocol_options) ->
    econf:and_then(
      econf:list(econf:binary()),
      fun(Options) -> str:join(Options, <<"|">>) end);
listen_opt_type(tls_compression) ->
    econf:bool();
listen_opt_type(tls) ->
    econf:bool();
listen_opt_type(max_stanza_size) ->
    econf:pos_int(infinity);
listen_opt_type(max_fsm_queue) ->
    econf:pos_int();
listen_opt_type(send_timeout) ->
    econf:timeout(second, infinity);
listen_opt_type(shaper) ->
    econf:shaper();
listen_opt_type(access) ->
    econf:acl();
listen_opt_type(use_proxy_protocol) ->
    econf:bool().

listen_options() ->
    [module, port,
     {transport, tcp},
     {ip, {0,0,0,0}},
     {accept_interval, 0},
     {send_timeout, 15000},
     {backlog, 5},
     {use_proxy_protocol, false},
     {supervisor, true}].
