path: root/lib/diameter/src/base/diameter_service.erl

%%
%% %CopyrightBegin%
%%
%% Copyright Ericsson AB 2010-2017. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%
%% %CopyrightEnd%
%%

%%
%% Implements the process that represents a service.
%%

-module(diameter_service).
-behaviour(gen_server).

%% towards diameter_service_sup
-export([start_link/1]).

%% towards diameter
-export([subscribe/1,
         unsubscribe/1,
         services/0,
         peer_info/1,
         info/2]).

%% towards diameter_config
-export([start/1,
         stop/1,
         start_transport/2,
         stop_transport/2]).

%% towards diameter_peer
-export([notify/2]).

%% towards diameter_traffic
-export([find_incoming_app/4,
         pick_peer/3]).

%% test/debug
-export([services/1,
         subscriptions/1,
         subscriptions/0,
         call_module/3,
         whois/1,
         state/1,
         uptime/1]).

%% gen_server callbacks
-export([init/1,
         handle_call/3,
         handle_cast/2,
         handle_info/2,
         terminate/2,
         code_change/3]).

-include_lib("diameter/include/diameter.hrl").
-include("diameter_internal.hrl").

%% RFC 3539 watchdog states.
-define(WD_INITIAL, initial).
-define(WD_OKAY,    okay).
-define(WD_SUSPECT, suspect).
-define(WD_DOWN,    down).
-define(WD_REOPEN,  reopen).

-type wd_state() :: ?WD_INITIAL
                  | ?WD_OKAY
                  | ?WD_SUSPECT
                  | ?WD_DOWN
                  | ?WD_REOPEN.

-define(DEFAULT_TC,     30000).  %% RFC 3588 ch 2.1
-define(RESTART_TC,      1000).  %% if restart was this recent

%% Maintain state in a table since a service's state is accessed
%% outside of the service process.
-define(STATE_TABLE, ?MODULE).

%% Workaround for dialyzer's lack of understanding of match specs.
-type match(T)
   :: T | '_' | '$1' | '$2'.

%% State of service gen_server. Note that the state term itself
%% doesn't change, which is relevant for the stateless application
%% callbacks since the state is retrieved from ?STATE_TABLE from
%% outside the service process. The pid in the service record is used
%% to determine whether or not we need to call the process for a
%% pick_peer callback in the statefull case.
-record(state,
        {id = diameter_lib:now(),
         service_name :: diameter:service_name(), %% key in ?STATE_TABLE
         service :: #diameter_service{},
         watchdogT = ets_new(watchdogs) %% #watchdog{} at start
                  :: ets:tid(),
         local  :: {ets:tid(), ets:tid(), ets:tid()},
         remote :: {ets:tid(), ets:tid(), ets:tid()},
         monitor = false :: false | pid(),   %% process to die with
         options :: #{sequence := diameter:sequence(),  %% sequence mask
                      share_peers := diameter:remotes(),%% broadcast to
                      use_shared_peers := diameter:remotes(),%% use from
                      restrict_connections := diameter:restriction(),
                      incoming_maxlen := diameter:message_length(),
                      strict_arities => diameter:strict_arities(),
                      strict_mbit := boolean(),
                      decode_format := diameter:decode_format(),
                      traffic_counters := boolean(),
                      string_decode := boolean(),
                      capabilities_cb => diameter:evaluable(),
                      pool_size => pos_integer(),
                      capx_timeout => diameter:'Unsigned32'(),
                      strict_capx => boolean(),
                      disconnect_cb => diameter:evaluable(),
                      dpr_timeout => diameter:'Unsigned32'(),
                      dpa_timeout => diameter:'Unsigned32'(),
                      length_errors => exit | handle | discard,
                      connect_timer => diameter:'Unsigned32'(),
                      watchdog_timer => diameter:'Unsigned32'()
                                      | {module(), atom(), list()},
                      watchdog_config => [{okay|suspect, non_neg_integer()}],
                      spawn_opt := list() | {module(), atom(), list()}}}).

%% Record representing an RFC 3539 watchdog process implemented by
%% diameter_watchdog.
-record(watchdog,
        {pid  :: match(pid()) | undefined,
         type :: match(connect | accept),
         ref  :: match(reference()),  %% key into diameter_config
         options :: match([diameter:transport_opt()]),%% from start_transport
         state = ?WD_INITIAL :: match(wd_state()),
         started = diameter_lib:now(),%% at process start
         peer = false :: match(boolean() | pid())}).
                      %% true at accepted/remove, pid() at okay/reopen

%% Record representing a Peer State Machine processes implemented by
%% diameter_peer_fsm.
-record(peer,
        {pid  :: pid(),
         apps :: match([{0..16#FFFFFFFF, diameter:app_alias()}] %% {Id, Alias}
                       | [diameter:app_alias()]),               %% remote
         caps :: match(#diameter_caps{}),
         started = diameter_lib:now(),     %% at process start or sharing
         watchdog :: match(pid()           %% key into watchdogT
                           | undefined)}). %% undefined if remote

%% ---------------------------------------------------------------------------
%% # start/1
%% ---------------------------------------------------------------------------

start(SvcName) ->
    diameter_service_sup:start_child(SvcName).

start_link(SvcName) ->
    Options = [{spawn_opt, diameter_lib:spawn_opts(server, [])}],
    gen_server:start_link(?MODULE, [SvcName], Options).
%% Put the arbitrary term SvcName in a list in case we ever want to
%% send more than this and need to distinguish old from new.

%% ---------------------------------------------------------------------------
%% # stop/1
%% ---------------------------------------------------------------------------

stop(SvcName) ->
    case whois(SvcName) of
        undefined ->
            {error, not_started};
        Pid ->
            stop(call_service(Pid, stop), Pid)
    end.

stop(ok, Pid) ->
    MRef = monitor(process, Pid),
    receive {'DOWN', MRef, process, _, _} -> ok end;
stop(No, _) ->
    No.

%% ---------------------------------------------------------------------------
%% # start_transport/3
%% ---------------------------------------------------------------------------

start_transport(SvcName, {_Ref, _Type, _Opts} = T) ->
    call_service_by_name(SvcName, {start, T}).

%% ---------------------------------------------------------------------------
%% # stop_transport/2
%% ---------------------------------------------------------------------------

stop_transport(_, []) ->
    ok;
stop_transport(SvcName, [_|_] = Refs) ->
    call_service_by_name(SvcName, {stop, Refs}).

%% ---------------------------------------------------------------------------
%% # info/2
%% ---------------------------------------------------------------------------

info(SvcName, Item) ->
    case lookup_state(SvcName) of
        [S] ->
            service_info(Item, S);
        [] ->
            undefined
    end.

%% lookup_state/1

lookup_state(SvcName) ->
    case ets:lookup(?STATE_TABLE, SvcName) of
        [#state{}] = L ->
            L;
        _ ->
            []
    end.

%% ---------------------------------------------------------------------------
%% # peer_info/2
%% ---------------------------------------------------------------------------

%% An extended version of info_peer/1 for peer_info/1.
peer_info(Pid) ->
    try
        {_, PD} = process_info(Pid, dictionary),
        {_, T} = lists:keyfind({diameter_peer_fsm, start}, 1, PD),
        {TPid, {{Type, Ref}, TMod, Cfg}} = T,
        {_, TD} = process_info(TPid, dictionary),
        {_, Data} = lists:keyfind({TMod, info}, 1, TD),
        [{ref, Ref},
         {type, Type},
         {owner, TPid},
         {module, TMod},
         {config, Cfg}
         | try TMod:info(Data) catch _:_ -> [] end]
    catch
        error:_ ->
            []
    end.

%% ---------------------------------------------------------------------------
%% # subscribe/1
%% # unsubscribe/1
%% ---------------------------------------------------------------------------

subscribe(SvcName) ->
    diameter_reg:add({?MODULE, subscriber, SvcName}).

unsubscribe(SvcName) ->
    diameter_reg:remove({?MODULE, subscriber, SvcName}).

subscriptions(Pat) ->
    pmap(diameter_reg:match({?MODULE, subscriber, Pat})).

subscriptions() ->
    subscriptions('_').

pmap(Props) ->
    lists:map(fun({{?MODULE, _, Name}, Pid}) -> {Name, Pid} end, Props).

%% ---------------------------------------------------------------------------
%% # services/1
%% ---------------------------------------------------------------------------

services(Pat) ->
    pmap(diameter_reg:match({?MODULE, service, Pat})).

services() ->
    services('_').

whois(SvcName) ->
    case diameter_reg:match({?MODULE, service, SvcName}) of
        [{_, Pid}] ->
            Pid;
        [] ->
            undefined
    end.

%% ---------------------------------------------------------------------------
%% # pick_peer/3
%% ---------------------------------------------------------------------------

-spec pick_peer(SvcName, AppOrAlias, Opts)
   -> {{{TPid, Caps}, App}, SvcOpts}
    | false  %% no selection
    | {error, no_service}
 when SvcName :: diameter:service_name(),
      AppOrAlias :: #diameter_app{}
                  | {alias, diameter:app_alias()},
      Opts :: {fun((Dict :: module()) -> [term()]),
               diameter:peer_filter(),
               Xtra :: list(),
               [diameter:peer_ref()]},
      TPid :: pid(),
      Caps :: #diameter_caps{},
      App  :: #diameter_app{},
      SvcOpts :: map().

pick_peer(SvcName, App, Opts) ->
    pick(lookup_state(SvcName), App, Opts).

pick([], _, _) ->
    {error, no_service};

pick([S], App, Opts) ->
    pick(S, App, Opts);

pick(#state{service = #diameter_service{applications = Apps}}
     = S,
     {alias, Alias},
     Opts) ->  %% initial call from diameter:call/4
    pick(S, find_outgoing_app(Alias, Apps), Opts);

pick(_, false = No, _) ->
    No;

pick(#state{options = SvcOpts}
     = S,
     #diameter_app{module = ModX, dictionary = Dict}
     = App0,
     {DestF, Filter, Xtra, TPids}) ->
    App = App0#diameter_app{module = ModX ++ Xtra},
    [_,_] = RealmAndHost = diameter_lib:eval([DestF, Dict]),
    case pick_peer(App, RealmAndHost, [Filter | TPids], S) of
        {_TPid, _Caps} = TC ->
            {{TC, App}, SvcOpts};
        false = No ->
            No
    end.

%% ---------------------------------------------------------------------------
%% # find_incoming_app/4
%% ---------------------------------------------------------------------------

-spec find_incoming_app(PeerT, TPid, Id, Apps)
   -> {#diameter_app{}, #diameter_caps{}}  %% connection and suitable app
    | #diameter_caps{}                     %% connection but no suitable app
    | false                                %% no connection
 when PeerT :: ets:tid(),
      TPid  :: pid(),
      Id    :: non_neg_integer(),
      Apps  :: [#diameter_app{}].

find_incoming_app(PeerT, TPid, Id, Apps) ->
    try ets:lookup(PeerT, TPid) of
        [#peer{} = P] ->
            find_incoming_app(P, Id, Apps);
        [] ->             %% transport has gone down
            false
    catch
        error: badarg ->  %% service has gone down (and taken table with it)
            false
    end.

%% ---------------------------------------------------------------------------
%% # notify/2
%% ---------------------------------------------------------------------------

notify(SvcName, Msg) ->
    Pid = whois(SvcName),
    is_pid(Pid) andalso (Pid ! Msg).

%% ===========================================================================
%% ===========================================================================

state(Svc) ->
    call_service(Svc, state).

uptime(Svc) ->
    call_service(Svc, uptime).

%% call_module/3

call_module(Service, AppMod, Request) ->
    call_service(Service, {call_module, AppMod, Request}).

%% ---------------------------------------------------------------------------
%% # init/1
%% ---------------------------------------------------------------------------

init([SvcName]) ->
    process_flag(trap_exit, true),  %% ensure terminate(shutdown, _)
    i(SvcName, diameter_reg:add_new({?MODULE, service, SvcName})).

i(SvcName, true) ->
    {ok, i(SvcName)};
i(_, false) ->
    {stop, {shutdown, already_started}}.

%% ---------------------------------------------------------------------------
%% # handle_call/3
%% ---------------------------------------------------------------------------

handle_call(state, _, S) ->
    {reply, S, S};

handle_call(uptime, _, #state{id = T} = S) ->
    {reply, diameter_lib:now_diff(T), S};

%% Start a transport.
handle_call({start, {Ref, Type, Opts}}, _From, S) ->
    {reply, start(Ref, {Type, Opts}, S), S};

%% Stop transports.
handle_call({stop, Refs}, _From, S) ->
    shutdown(Refs, S),
    {reply, ok, S};

%% pick_peer with mutable state
handle_call({pick_peer, Local, Remote, App}, _From, S) ->
    #diameter_app{mutable = true} = App,  %% assert
    {reply, pick_peer(Local, Remote, self(), S#state.service_name, App), S};

handle_call({call_module, AppMod, Req}, From, S) ->
    call_module(AppMod, Req, From, S);

handle_call(stop, _From, S) ->
    shutdown(service, S),
    {stop, normal, ok, S};
%% The server currently isn't guaranteed to be dead when the caller
%% gets the reply. We deal with this in the call to the server,
%% stating a monitor that waits for DOWN before returning.

handle_call(Req, From, S) ->
    unexpected(handle_call, [Req, From], S),
    {reply, nok, S}.

%% ---------------------------------------------------------------------------
%% # handle_cast/2
%% ---------------------------------------------------------------------------

handle_cast(Req, S) ->
    unexpected(handle_cast, [Req], S),
    {noreply, S}.

%% ---------------------------------------------------------------------------
%% # handle_info/2
%% ---------------------------------------------------------------------------

handle_info(T, #state{} = S) ->
    case transition(T,S) of
        ok ->
            {noreply, S};
        {stop, Reason} ->
            {stop, {shutdown, Reason}, S}
    end.

%% transition/2

%% Peer process is telling us to start a new accept process.
transition({accepted, Pid, TPid}, S) ->
    accepted(Pid, TPid, S),
    ok;

%% Connecting transport is being restarted by watchdog.
transition({reconnect, Pid}, S) ->
    reconnect(Pid, S),
    ok;

%% Watchdog is sending notification of transport death.
transition({close, Pid, Reason}, #state{service_name = SvcName,
                                        watchdogT = WatchdogT}) ->
    #watchdog{state = WS,
              ref = Ref,
              type = Type,
              options = Opts}
        = fetch(WatchdogT, Pid),
    WS /= ?WD_OKAY
        andalso
        send_event(SvcName, {closed, Ref, Reason, {type(Type), Opts}}),
    ok;

%% Watchdog is sending notification of a state transition.
transition({watchdog, Pid, {[TPid | Data], From, To}},
           #state{service_name = SvcName,
                  watchdogT = WatchdogT}
           = S) ->
    #watchdog{ref = Ref, type = T, options = Opts}
        = Wd
        = fetch(WatchdogT, Pid),
    watchdog(TPid, Data, From, To, Wd, S),
    send_event(SvcName, {watchdog, Ref, TPid, {From, To}, {T, Opts}}),
    ok;
%% Death of a watchdog process (#watchdog.pid) results in the removal of
%% it's peer and any associated conn record when 'DOWN' is received.
%% Death of a peer process process (#peer.pid, #watchdog.peer) results in
%% ?WD_DOWN.

%% Monitor process has died. Just die with a reason that tells
%% diameter_config about the happening. If a cleaner shutdown is
%% required then someone should stop us.
transition({'DOWN', MRef, process, _, Reason}, #state{monitor = MRef}) ->
    {stop, {monitor, Reason}};

%% Local watchdog process has died.
transition({'DOWN', _, process, Pid, _Reason}, S)
  when node(Pid) == node() ->
    watchdog_down(Pid, S),
    ok;

%% Remote service wants to know about shared peers.
transition({service, Pid}, S) ->
    share_peers(Pid, S),
    ok;

%% Remote service is communicating a shared peer.
transition({peer, TPid, Aliases, Caps}, S) ->
    remote_peer_up(TPid, Aliases, Caps, S),
    ok;
transition({peer, TPid}, S) ->
    remote_peer_down(TPid, S),
    ok;

%% Remote peer process has died.
transition({'DOWN', _, process, TPid, _}, S) ->
    remote_peer_down(TPid, S),
    ok;

%% Restart after tc expiry.
transition({tc_timeout, T}, S) ->
    tc_timeout(T, S),
    ok;

transition({nodeup, Node, _}, S) ->
    nodeup(Node, S),
    ok;

transition({nodedown, _Node, _}, _) ->
    ok;

transition(Req, S) ->
    unexpected(handle_info, [Req], S),
    ok.

%% ---------------------------------------------------------------------------
%% # terminate/2
%% ---------------------------------------------------------------------------

terminate(Reason, #state{service_name = Name, local = {PeerT, _, _}} = S) ->
    send_event(Name, stop),
    ets:delete(?STATE_TABLE, Name),

    %% Communicate pending loss of any peers that connection_down/3
    %% won't. This is needed when stopping a service since we don't
    %% wait for watchdog state changes to take care of if. That this
    %% takes place after deleting the state entry ensures that the
    %% resulting failover by request processes accomplishes nothing.
    ets:foldl(fun(#peer{pid = TPid}, _) ->
                      diameter_traffic:peer_down(TPid)
              end,
              ok,
              PeerT),

    shutdown == Reason  %% application shutdown
        andalso shutdown(application, S).

%% ---------------------------------------------------------------------------
%% # code_change/3
%% ---------------------------------------------------------------------------

code_change(_FromVsn, S, _Extra) ->
    {ok, S}.

%% ===========================================================================
%% ===========================================================================

unexpected(F, A, #state{service_name = Name}) ->
    ?UNEXPECTED(F, A ++ [Name]).

eval([M|X], F, A) ->
    apply(M, F, A ++ X).

%% Callback with state.

state_cb(#diameter_app{module = ModX, mutable = false, init_state = S},
         pick_peer = F,
         A) ->
    eval(ModX, F, A ++ [S]);

state_cb(#diameter_app{module = ModX, alias = Alias}, F, A) ->
    eval(ModX, F, A ++ [mod_state(Alias)]).

choose(true, X, _)  -> X;
choose(false, _, X) -> X.

ets_new(Tbl) ->
    ets:new(Tbl, [{keypos, 2}]).

%% Using the process dictionary for the callback state was initially
%% just a way to make what was horrendous trace (big state record and
%% much else everywhere) somewhat more readable. There's not as much
%% need for it now but it's no worse (except possibly that we don't
%% see the table identifier being passed around) than an ets table so
%% keep it.

mod_state(Alias) ->
    get({?MODULE, mod_state, Alias}).

mod_state(Alias, ModS) ->
    put({?MODULE, mod_state, Alias}, ModS).

%% ---------------------------------------------------------------------------
%% # shutdown/2
%% ---------------------------------------------------------------------------

%% remove_transport
shutdown(Refs, #state{watchdogT = WatchdogT})
  when is_list(Refs) ->
    ets:insert(WatchdogT, ets:foldl(fun(R,A) -> st(R, Refs, A) end,
                                    [],
                                    WatchdogT));

%% application/service shutdown
shutdown(Reason, #state{watchdogT = WatchdogT})
  when Reason == application;
       Reason == service ->
    diameter_lib:wait(ets:foldl(fun(P,A) -> ss(P, Reason, A) end,
                                [],
                                WatchdogT)).

%% st/3

%% Mark replacement as started so that a subsequent accept doesn't
%% result in a new process that isn't terminated.
st(#watchdog{ref = Ref, pid = Pid, peer = P} = Rec, Refs, Acc) ->
    case lists:member(Ref, Refs) of
        true ->
            Pid ! {shutdown, self(), transport},  %% 'DOWN' cleans up
            [Rec#watchdog{peer = true} || P == false] ++ Acc;
        false ->
            Acc
    end.

%% ss/3

ss(#watchdog{pid = Pid}, Reason, Acc) ->
    MRef = monitor(process, Pid),
    Pid ! {shutdown, self(), Reason},
    [MRef | Acc].

%% ---------------------------------------------------------------------------
%% # call_service/2
%% ---------------------------------------------------------------------------

call_service(Pid, Req)
  when is_pid(Pid) ->
    cs(Pid, Req);
call_service(SvcName, Req) ->
    call_service_by_name(SvcName, Req).

call_service_by_name(SvcName, Req) ->
    cs(whois(SvcName), Req).

cs(Pid, Req)
  when is_pid(Pid) ->
    try
        gen_server:call(Pid, Req, infinity)
    catch
        E: Reason when E == exit ->
            {error, {E, Reason}}
    end;

cs(undefined, _) ->
    {error, no_service}.

%% ---------------------------------------------------------------------------
%% # i/1
%% ---------------------------------------------------------------------------

%% Intialize the state of a service gen_server.

i(SvcName) ->
    %% Split the config into a server state and a list of transports.
    {#state{} = S, CL} = lists:foldl(fun cfg_acc/2,
                                     {false, []},
                                     diameter_config:lookup(SvcName)),

    %% Publish the state in order to be able to access it outside of
    %% the service process. Originally table identifiers were only
    %% known to the service process but we now want to provide the
    %% option of application callbacks being 'stateless' in order to
    %% avoid having to go through a common process. (Eg. An agent that
    %% sends a request for every incoming request.)
    true = ets:insert_new(?STATE_TABLE, S),

    %% Start fsms for each transport.
    send_event(SvcName, start),
    lists:foreach(fun(T) -> start_fsm(T,S) end, CL),

    init_shared(S),
    S.

cfg_acc({SvcName, #diameter_service{applications = Apps} = Rec, Opts},
        {false, Acc}) ->
    lists:foreach(fun init_mod/1, Apps),
    #{monitor := M}
        = SvcOpts
        = service_opts(Opts),
    S = #state{service_name = SvcName,
               service = Rec#diameter_service{pid = self()},
               local   = init_peers(),
               remote  = init_peers(),
               monitor = mref(M),
               options = maps:remove(monitor, SvcOpts)},
    {S, Acc};

cfg_acc({_Ref, Type, _Opts} = T, {S, Acc})
  when Type == connect;
       Type == listen ->
    {S, [T | Acc]}.

init_peers() ->
    {ets_new(caps),            %% #peer{}
     ets:new(apps, [bag]),     %% {Alias, TPid}
     ets:new(idents, [bag])}.  %% {{host, OH} | {realm, OR} | {OR, OH},
                               %%  Alias,
                               %%  TPid}

service_opts(Opts) ->
    remove([{strict_arities, true}],
           maps:merge(maps:from_list([{monitor, false} | def_opts()]),
                      maps:from_list(Opts))).

remove(List, Map) ->
    maps:filter(fun(K,V) -> not lists:member({K,V}, List) end,
                Map).

def_opts() ->  %% defaults on the service map
    [{share_peers, false},
     {use_shared_peers, false},
     {sequence, {0,32}},
     {restrict_connections, nodes},
     {incoming_maxlen, 16#FFFFFF},
     {strict_arities, true},
     {strict_mbit, true},
     {decode_format, record},
     {traffic_counters, true},
     {string_decode, true},
     {spawn_opt, []}].

mref(false = No) ->
    No;
mref(P) ->
    monitor(process, P).

init_shared(#state{options = #{use_shared_peers := T},
                   service_name = Svc}) ->
    T == false orelse net_kernel:monitor_nodes(true, [{node_type, visible},
                                                      nodedown_reason]),
    notify(T, Svc, {service, self()}).

init_mod(#diameter_app{alias = Alias,
                       init_state = S}) ->
    mod_state(Alias, S).

start_fsm({Ref, Type, Opts}, S) ->
    start(Ref, {Type, Opts}, S).

notify(Share, SvcName, T) ->
    Nodes = remotes(Share),
    [] /= Nodes andalso diameter_peer:notify(Nodes, SvcName, T).
%% Test for the empty list for upgrade reasons: there's no
%% diameter_peer:notify/3 in old code.

nodeup(Node, #state{options = #{share_peers := SP},
                    service_name = SvcName}) ->
    lists:member(Node, remotes(SP))
        andalso diameter_peer:notify([Node], SvcName, {service, self()}).

remotes(false) ->
    [];

remotes(true) ->
    nodes();

remotes(Nodes)
  when is_atom(hd(Nodes));
       Nodes == [] ->
    Nodes;

remotes(F) ->
    try diameter_lib:eval(F) of
        L when is_list(L) ->
            L;
        T ->
            ?LOG(invalid_return, {F,T}),
            error_report(invalid_return, share_peers, F),
            []
    catch
        E:R ->
            ?LOG(failure, {E, R, F, diameter_lib:get_stacktrace()}),
            error_report(failure, share_peers, F),
            []
    end.

%% error_report/3

error_report(T, What, F) ->
    Reason = io_lib:format("~s from ~p callback", [reason(T), What]),
    diameter_lib:error_report(Reason, diameter_lib:eval_name(F)).

reason(invalid_return) ->
    "invalid return";
reason(failure) ->
    "failure".

%% ---------------------------------------------------------------------------
%% # start/3
%% ---------------------------------------------------------------------------

%% If the initial start/3 at service/transport start succeeds then
%% subsequent calls to start/4 on the same service will also succeed
%% since they involve the same call to merge_service/2. We merge here
%% rather than earlier since the service may not yet be configured
%% when the transport is configured.

start(Ref, {T, Opts}, S)
  when T == connect;
       T == listen ->
    N = proplists:get_value(pool_size, Opts, 1),
    try
        {ok, start(Ref, type(T), Opts, N, S)}
    catch
        ?FAILURE(Reason) ->
            {error, Reason}
    end.
%% TODO: don't actually raise any errors yet

%% There used to be a difference here between the handling of
%% configured listening and connecting transports but now we simply
%% tell the transport_module to start an accepting or connecting
%% process respectively, the transport implementation initiating
%% listening on a port as required.
type(listen)      -> accept;
type(accept)      -> listen;
type(connect = T) -> T.

%% start/4

start(Ref, Type, Opts, State) ->
    start(Ref, Type, Opts, 1, State).

%% start/5

start(Ref, Type, Opts, N, #state{watchdogT = WatchdogT,
                                 local = {PeerT, _, _},
                                 options = #{string_decode := SD}
                                 = SvcOpts,
                                 service_name = SvcName,
                                 service = Svc0})
  when Type == connect;
       Type == accept ->
    #diameter_service{applications = Apps}
        = Svc1
        = merge_service(Opts, Svc0),
    Svc = binary_caps(Svc1, SD),
    {SOpts, TOpts} = merge_opts(SvcOpts, Opts),
    RecvData = diameter_traffic:make_recvdata([SvcName, PeerT, Apps, SOpts]),
    T = {TOpts, SOpts, RecvData, Svc},
    Rec = #watchdog{type = Type,
                    ref = Ref,
                    options = TOpts},

    diameter_lib:fold_n(fun(_,A) ->
                                [wd(Type, Ref, T, WatchdogT, Rec) | A]
                        end,
                        [],
                        N).

merge_opts(SvcOpts, Opts) ->
    Keys = [K || {K,_} <- def_opts()],
    SO = [T || {K,_} = T <- Opts, lists:member(K, Keys)],
    TO = Opts -- SO,
    {maps:merge(maps:with(Keys, SvcOpts), maps:from_list(SO)),
     TO ++ [T || {K,_} = T <- maps:to_list(SvcOpts),
                 not lists:member(K, Keys),
                 not lists:keymember(K, 1, Opts)]}.

binary_caps(Svc, true) ->
    Svc;
binary_caps(#diameter_service{capabilities = Caps} = Svc, false) ->
    Svc#diameter_service{capabilities = diameter_capx:binary_caps(Caps)}.

wd(Type, Ref, T, WatchdogT, Rec) ->
    Pid = start_watchdog(Type, Ref, T),
    ets:insert(WatchdogT, Rec#watchdog{pid = Pid}),
    Pid.

%% Note that the service record passed into the watchdog is the merged
%% record so that each watchdog may get a different record. This
%% record is what is passed back into application callbacks.

start_watchdog(Type, Ref, T) ->
    {_MRef, Pid} = diameter_watchdog:start({Type, Ref}, T),
    Pid.

%% merge_service/2

merge_service(Opts, Svc) ->
    lists:foldl(fun ms/2, Svc, Opts).

%% Limit the applications known to the fsm to those in the 'apps'
%% option. That this might be empty is checked by the fsm. It's not
%% checked at config-time since there's no requirement that the
%% service be configured first. (Which could be considered a bit odd.)
ms({applications, As}, #diameter_service{applications = Apps} = S)
  when is_list(As) ->
    S#diameter_service{applications
                       = [A || A <- Apps,
                               lists:member(A#diameter_app.alias, As)]};

%% The fact that all capabilities can be configured on the transports
%% means that the service doesn't necessarily represent a single
%% locally implemented Diameter node as identified by Origin-Host: a
%% transport can configure its own Origin-Host. This means that the
%% service little more than a placeholder for default capabilities
%% plus a list of applications that individual transports can choose
%% to support (or not).
ms({capabilities, Opts}, #diameter_service{capabilities = Caps0} = Svc)
  when is_list(Opts) ->
    %% make_caps has already succeeded in diameter_config so it will succeed
    %% again here.
    {ok, Caps} = diameter_capx:make_caps(Caps0, Opts),
    Svc#diameter_service{capabilities = Caps};

ms(_, Svc) ->
    Svc.

%% ---------------------------------------------------------------------------
%% # accepted/3
%% ---------------------------------------------------------------------------

accepted(Pid, _TPid, #state{watchdogT = WatchdogT} = S) ->
    #watchdog{type = accept = T, peer = P}
        = Wd
        = fetch(WatchdogT, Pid),
    if not P ->
            #watchdog{ref = Ref, options = Opts} = Wd,
            %% Mark replacement started, and start new watchdog.
            ets:insert(WatchdogT, Wd#watchdog{peer = true}),
            start(Ref, T, Opts, S);
       P ->
            %% Transport removal in progress: true has been set in
            %% shutdown/2, and the transport will die as a
            %% consequence.
            ok
    end.

%% fetch/2

fetch(Tid, Key) ->
    [T] = ets:lookup(Tid, Key),
    T.

%% ---------------------------------------------------------------------------
%% # watchdog/6
%%
%% React to a watchdog state transition.
%% ---------------------------------------------------------------------------

%% Watchdog has a new open connection.
watchdog(TPid, [T], _, ?WD_OKAY, Wd, State) ->
    connection_up({TPid, T}, Wd, State);

%% Watchdog has a new connection that will be opened after DW[RA]
%% exchange.
watchdog(TPid, [T], _, ?WD_REOPEN, Wd, State) ->
    reopen({TPid, T}, Wd, State);

%% Watchdog has recovered a suspect connection.
watchdog(TPid, [], ?WD_SUSPECT, ?WD_OKAY, Wd, State) ->
    #watchdog{peer = TPid} = Wd,  %% assert
    connection_up(Wd, State);

%% Watchdog has an unresponsive connection. Note that the peer table
%% entry isn't removed until DOWN.
watchdog(TPid, [], ?WD_OKAY, ?WD_SUSPECT = To, Wd, State) ->
    #watchdog{peer = TPid} = Wd,  %% assert
    watchdog_down(Wd, To, State);

%% Watchdog has lost its connection.
watchdog(TPid, [], _, ?WD_DOWN = To, Wd, #state{local = {PeerT, _, _}} = S) ->
    close(Wd),
    watchdog_down(Wd, To, S),
    ets:delete(PeerT, TPid);

watchdog(_, [], _, _, _, _) ->
    ok.

watchdog_down(Wd, To, #state{watchdogT = WatchdogT} = S) ->
    ets:insert(WatchdogT, Wd#watchdog{state = To}),
    connection_down(Wd, To, S).

%% ---------------------------------------------------------------------------
%% # connection_up/3
%% ---------------------------------------------------------------------------

%% Watchdog process has reached state OKAY.

connection_up({TPid, {Caps, SupportedApps, Pkt}},
              #watchdog{pid = Pid}
              = Wd,
              #state{local = {PeerT, _, _}}
              = S) ->
    Rec = #peer{pid = TPid,
                apps = SupportedApps,
                caps = Caps,
                watchdog = Pid},
    ets:insert(PeerT, Rec),
    connection_up([Pkt], Wd#watchdog{peer = TPid}, Rec, S).

%% ---------------------------------------------------------------------------
%% # reopen/3
%% ---------------------------------------------------------------------------

reopen({TPid, {Caps, SupportedApps, _Pkt}},
       #watchdog{pid = Pid}
       = Wd,
       #state{watchdogT = WatchdogT,
              local = {PeerT, _, _}}) ->
    ets:insert(PeerT, #peer{pid = TPid,
                            apps = SupportedApps,
                            caps = Caps,
                            watchdog = Pid}),
    ets:insert(WatchdogT, Wd#watchdog{state = ?WD_REOPEN,
                                      peer = TPid}).

%% ---------------------------------------------------------------------------
%% # connection_up/2
%% ---------------------------------------------------------------------------

%% Watchdog has recovered a suspect connection. Note that there has
%% been no new capabilties exchange in this case.

connection_up(#watchdog{peer = TPid} = Wd, #state{local = {PeerT, _, _}}
                                           = S) ->
    connection_up([], Wd, fetch(PeerT, TPid), S).

%% connection_up/4

connection_up(Extra,
              #watchdog{peer = TPid}
              = Wd,
              #peer{apps = SApps, caps = Caps}
              = Pr,
              #state{watchdogT = WatchdogT,
                     local = LT,
                     service_name = SvcName,
                     service = #diameter_service{applications = Apps}}
              = S) ->
    ets:insert(WatchdogT, Wd#watchdog{state = ?WD_OKAY}),
    diameter_traffic:peer_up(TPid),
    local_peer_up(SApps, {TPid, Caps}, {SvcName, Apps}, LT),
    report_status(up, Wd, Pr, S, Extra).

local_peer_up(SApps, {TPid, Caps} = TC, SA, LT) ->
    insert_peer(TPid, [A || {_,A} <- SApps], Caps, LT),
    lists:foreach(fun(A) -> peer_up(A, TC, SA) end, SApps).

peer_up({Id, Alias}, TC, SA) ->
    peer_up(Id, Alias, TC, SA).

peer_up(Id, Alias, {TPid, _} = TC, {SvcName, Apps}) ->
    #diameter_app{id = Id}  %% assert
        = App
        = find_app(Alias, Apps),

    peer_cb(App, peer_up, [SvcName, TC])
        orelse exit(TPid, kill).  %% fake transport failure

%% ---------------------------------------------------------------------------
%% # find_incoming_app/3
%% ---------------------------------------------------------------------------

%% No one should be sending the relay identifier.
find_incoming_app(#peer{caps = Caps}, ?APP_ID_RELAY, _) ->
    Caps;

find_incoming_app(Peer, Id, Apps)
  when is_integer(Id) ->
    find_incoming_app(Peer, [Id, ?APP_ID_RELAY], Apps);

%% Note that the apps represented in SApps may be a strict subset of
%% those in Apps.
find_incoming_app(#peer{apps = SApps, caps = Caps}, Ids, Apps) ->
    case keyfind(Ids, 1, SApps) of
        {_Id, Alias} ->
            {#diameter_app{} = find_app(Alias, Apps), Caps};
        false ->
            Caps
    end.

%% keyfind/3

keyfind([], _, _) ->
    false;
keyfind([Key | Rest], Pos, L) ->
    case lists:keyfind(Key, Pos, L) of
        false ->
            keyfind(Rest, Pos, L);
        T ->
            T
    end.

%% find_outgoing_app/2

find_outgoing_app(Alias, Apps) ->
    case find_app(Alias, Apps) of
        #diameter_app{id = ?APP_ID_RELAY} ->
            false;
        A ->
            A
    end.

%% find_app/2

find_app(Alias, Apps) ->
    lists:keyfind(Alias, #diameter_app.alias, Apps).

%% Don't bring down the service (and all associated connections)
%% regardless of what happens.
peer_cb(App, F, A) ->
    try state_cb(App, F, A) of
        ModS ->
            mod_state(App#diameter_app.alias, ModS),
            true
    catch
        E:R ->
            %% Don't include arguments since a #diameter_caps{} strings
            %% from the peer, which could be anything (especially, large).
            [Mod|X] = App#diameter_app.module,
            ?LOG(failure, {E, R, Mod, F, diameter_lib:get_stacktrace()}),
            error_report(failure, F, {Mod, F, A ++ X}),
            false
    end.

%% ---------------------------------------------------------------------------
%% # connection_down/3
%% ---------------------------------------------------------------------------

connection_down(#watchdog{state = ?WD_OKAY,
                          peer = TPid}
                = Wd,
                #peer{caps = Caps,
                      apps = SApps}
                = Pr,
                #state{service_name = SvcName,
                       service = #diameter_service{applications = Apps},
                       local = LT}
                = S) ->
    report_status(down, Wd, Pr, S, []),
    local_peer_down(SApps, {TPid, Caps}, {SvcName, Apps}, LT),
    diameter_traffic:peer_down(TPid);

connection_down(#watchdog{state = ?WD_OKAY,
                          peer = TPid}
                = Wd,
                To,
                #state{local = {PeerT, _, _}}
                = S)
  when is_atom(To) ->
    connection_down(Wd, #peer{} = fetch(PeerT, TPid), S);

connection_down(#watchdog{}, _, _) ->
    ok.

local_peer_down(SApps, {TPid, _Caps} = TC, SA, LT) ->
    delete_peer(TPid, LT),
    lists:foreach(fun(A) -> peer_down(A, TC, SA) end, SApps).

peer_down({Id, Alias}, TC, SA) ->
    peer_down(Id, Alias, TC, SA).

peer_down(Id, Alias, TC, {SvcName, Apps}) ->
    #diameter_app{id = Id}  %% assert
        = App
        = find_app(Alias, Apps),

    peer_cb(App, peer_down, [SvcName, TC]).

%% ---------------------------------------------------------------------------
%% # watchdog_down/2
%% ---------------------------------------------------------------------------

%% Watchdog process has died.

watchdog_down(Pid, #state{watchdogT = WatchdogT} = S) ->
    Wd = fetch(WatchdogT, Pid),
    ets:delete_object(WatchdogT, Wd),
    restart(Wd,S),
    wd_down(Wd,S).

%% Watchdog has never reached OKAY ...
wd_down(#watchdog{peer = B}, _)
  when is_boolean(B) ->
    ok;

%% ... or maybe it has.
wd_down(#watchdog{peer = TPid} = Wd, #state{local = {PeerT, _, _}} = S) ->
    connection_down(Wd, ?WD_DOWN, S),
    ets:delete(PeerT, TPid).

%% restart/2

restart(Wd, S) ->
    q_restart(restart(Wd), S).

%% restart/1

%% Always try to reconnect.
restart(#watchdog{ref = Ref,
                  type = connect = T,
                  options = Opts,
                  started = Time}) ->
    {Time, {Ref, T, Opts}};

%% Transport connection hasn't yet been accepted ...
restart(#watchdog{ref = Ref,
                  type = accept = T,
                  options = Opts,
                  peer = false,
                  started = Time}) ->
    {Time, {Ref, T, Opts}};

%% ... or it has: a replacement has already been spawned.
restart(#watchdog{type = accept}) ->
    false.

%% q_restart/2

%% Start the reconnect timer.
q_restart({Time, {_Ref, Type, Opts} = T}, S) ->
    start_tc(tc(Time, default_tc(Type, Opts)), T, S);
q_restart(false, _) ->
    ok.

%% RFC 3588, 2.1:
%%
%%   When no transport connection exists with a peer, an attempt to
%%   connect SHOULD be periodically made.  This behavior is handled via
%%   the Tc timer, whose recommended value is 30 seconds.  There are
%%   certain exceptions to this rule, such as when a peer has terminated
%%   the transport connection stating that it does not wish to
%%   communicate.

default_tc(connect, Opts) ->
    connect_timer(Opts, ?DEFAULT_TC);
default_tc(accept, _) ->
    0.

%% Accept both connect_timer and the (older) reconnect_timer, the
%% latter being a remnant from a time in which the timer did apply to
%% reconnect attempts.
connect_timer(Opts, Def0) ->
    Def = proplists:get_value(reconnect_timer, Opts, Def0),
    proplists:get_value(connect_timer, Opts, Def).

%% Bound tc below if the watchdog was restarted recently to avoid
%% continuous restarted in case of faulty config or other problems.
tc(Time, Tc) ->
    choose(Tc > ?RESTART_TC
             orelse diameter_lib:micro_diff(Time) > 1000*?RESTART_TC,
           Tc,
           ?RESTART_TC).

start_tc(0, T, S) ->
    tc_timeout(T, S);
start_tc(Tc, T, _) ->
    erlang:send_after(Tc, self(), {tc_timeout, T}).

%% tc_timeout/2

tc_timeout({Ref, _Type, _Opts} = T, #state{service_name = SvcName} = S) ->
    tc(diameter_config:have_transport(SvcName, Ref), T, S).

tc(true, {Ref, Type, Opts}, #state{service_name = SvcName} = S) ->
    send_event(SvcName, {reconnect, Ref, Opts}),
    start(Ref, Type, Opts, S);
tc(false = No, _, _) ->  %% removed
    No.

%% ---------------------------------------------------------------------------
%% # close/2
%% ---------------------------------------------------------------------------

%% The watchdog doesn't start a new fsm in the accept case, it
%% simply stays alive until someone tells it to die in order for
%% another watchdog to be able to detect that it should transition
%% from initial into reopen rather than okay. That someone is either
%% the accepting watchdog upon reception of a CER from the previously
%% connected peer, or us after connect_timer timeout or immediately.

close(#watchdog{type = connect}) ->
    ok;

close(#watchdog{type = accept,
                pid = Pid,
                options = Opts}) ->
    Tc = connect_timer(Opts, 2*?DEFAULT_TC),
    erlang:send_after(Tc, Pid, close).
%% The RFC's only document the behaviour of Tc, our connect_timer,
%% for the establishment of connections but we also give
%% connect_timer semantics for a listener, being the time within
%% which a new connection attempt is expected of a connecting peer.
%% The value should be greater than the peer's Tc + jitter.

%% ---------------------------------------------------------------------------
%% # reconnect/2
%% ---------------------------------------------------------------------------

reconnect(Pid, #state{service_name = SvcName,
                      watchdogT = WatchdogT}) ->
    #watchdog{ref = Ref,
              type = connect,
              options = Opts}
        = fetch(WatchdogT, Pid),
    send_event(SvcName, {reconnect, Ref, Opts}).

%% ---------------------------------------------------------------------------
%% # call_module/4
%% ---------------------------------------------------------------------------

%% Backwards compatibility and never documented/advertised. May be
%% removed.

call_module(Mod, Req, From, #state{service
                                   = #diameter_service{applications = Apps},
                                   service_name = Svc}
                            = S) ->
    case cm([A || A <- Apps, Mod == hd(A#diameter_app.module)],
            Req,
            From,
            Svc)
    of
        {reply = T, RC} ->
            {T, RC, S};
        noreply = T ->
            {T, S};
        Reason ->
            {reply, {error, Reason}, S}
    end.

cm([#diameter_app{alias = Alias} = App], Req, From, Svc) ->
    Args = [Req, From, Svc],

    try state_cb(App, handle_call, Args) of
        {noreply = T, ModS} ->
            mod_state(Alias, ModS),
            T;
        {reply = T, RC, ModS} ->
            mod_state(Alias, ModS),
            {T, RC};
        T ->
            ModX = App#diameter_app.module,
            ?LOG(invalid_return, {ModX, handle_call, Args, T}),
            invalid
    catch
        E: Reason ->
            ModX = App#diameter_app.module,
            Stack = diameter_lib:get_stacktrace(),
            ?LOG(failure, {E, Reason, ModX, handle_call, Stack}),
            failure
    end;

cm([], _, _, _) ->
    unknown;

cm([_,_|_], _, _, _) ->
    multiple.

%% ---------------------------------------------------------------------------
%% # report_status/5
%% ---------------------------------------------------------------------------

report_status(Status,
              #watchdog{ref = Ref,
                        peer = TPid,
                        type = Type,
                        options = Opts},
              #peer{apps = [_|_] = Apps,
                    caps = Caps},
              #state{service_name = SvcName}
              = S,
              Extra) ->
    share_peer(Status, Caps, Apps, TPid, S),
    Info = [Status, Ref, {TPid, Caps}, {type(Type), Opts} | Extra],
    send_event(SvcName, list_to_tuple(Info)).

%% send_event/2

send_event(SvcName, Info) ->
    send_event(#diameter_event{service = SvcName,
                               info = Info}).

send_event(#diameter_event{service = SvcName} = E) ->
    lists:foreach(fun({_, Pid}) -> Pid ! E end, subscriptions(SvcName)).

%% ---------------------------------------------------------------------------
%% # share_peer/5
%% ---------------------------------------------------------------------------

share_peer(up, Caps, Apps, TPid, #state{options = #{share_peers := SP},
                                        service_name = Svc}) ->
    notify(SP, Svc, {peer, TPid, [A || {_,A} <- Apps], Caps});

share_peer(down, _Caps, _Apps, TPid, #state{options = #{share_peers := SP},
                                            service_name = Svc}) ->
    notify(SP, Svc, {peer, TPid}).

%% ---------------------------------------------------------------------------
%% # share_peers/2
%% ---------------------------------------------------------------------------

share_peers(Pid, #state{options = #{share_peers := SP},
                        local = {PeerT, AppT, _}}) ->
    is_remote(Pid, SP)
        andalso ets:foldl(fun(T, N) -> N + sp(Pid, AppT, T) end,
                          0,
                          PeerT).

%% An entry in the peer table doesn't mean the watchdog state is OKAY,
%% an entry in the app table does.

sp(Pid, AppT, #peer{pid = TPid,
                    apps = [{_, Alias} | _] = Apps,
                    caps = Caps}) ->
    Spec = [{{'$1', TPid},
             [{'==', '$1', {const, Alias}}],
             ['$_']}],
    case ets:select(AppT, Spec, 1) of
        '$end_of_table' ->
            0;
        _ ->
            Pid ! {peer, TPid, [A || {_,A} <- Apps], Caps},
            1
    end.

is_remote(Pid, T) ->
    Node = node(Pid),
    Node /= node() andalso lists:member(Node, remotes(T)).

%% ---------------------------------------------------------------------------
%% # remote_peer_up/4
%% ---------------------------------------------------------------------------

remote_peer_up(TPid, Aliases, Caps, #state{options = #{use_shared_peers := T},
                                           remote = {PeerT, _, _}}
                                    = S) ->
    is_remote(TPid, T)
        andalso not ets:member(PeerT, TPid)
        andalso rpu(TPid, Aliases, Caps, S).

%% Notification can be duplicate since remote nodes push and the local
%% node pulls.

rpu(TPid, Aliases, Caps, #state{service = Svc, remote = RT}) ->
    #diameter_service{applications = Apps} = Svc,
    Key = #diameter_app.alias,
    F = fun(A) -> lists:keymember(A, Key, Apps) end,
    rpu(TPid, lists:filter(F, Aliases), Caps, RT);

rpu(_, [] = No, _, _) ->
    No;

rpu(TPid, Aliases, Caps, {PeerT, _, _} = RT) ->
    monitor(process, TPid),
    ets:insert(PeerT, #peer{pid = TPid,
                            apps = Aliases,
                            caps = Caps}),
    insert_peer(TPid, Aliases, Caps, RT).

%% insert_peer/4

insert_peer(TPid, Aliases, Caps, {_PeerT, AppT, IdentT}) ->
    #diameter_caps{origin_host = {_, OH},
                   origin_realm = {_, OR}}
        = Caps,
    ets:insert(AppT, [{A, TPid} || A <- Aliases]),
    H = iolist_to_binary(OH),
    R = iolist_to_binary(OR),
    ets:insert(IdentT, [{T, A, TPid} || T <- [{host, H}, {realm, R}, {R, H}],
                                        A <- Aliases]).

%% ---------------------------------------------------------------------------
%% # remote_peer_down/2
%% ---------------------------------------------------------------------------

remote_peer_down(TPid, #state{remote = {PeerT, _, _} = RT}) ->
    ets:delete(PeerT, TPid),
    delete_peer(TPid, RT).

%% delete_peer/2

delete_peer(TPid, {_PeerT, AppT, IdentT}) ->
    ets:select_delete(AppT, [{{'_', TPid}, [], [true]}]),
    ets:select_delete(IdentT, [{{'_', '_', TPid}, [], [true]}]).

%% ---------------------------------------------------------------------------
%% pick_peer/4
%% ---------------------------------------------------------------------------

pick_peer(#diameter_app{alias = Alias}
          = App,
          RealmAndHost,
          Filter,
          #state{local = LT,
                 remote = RT,
                 service_name = SvcName,
                 service = #diameter_service{pid = Pid}}) ->
    pick_peer(peers(Alias, RealmAndHost, Filter, LT),
              peers(Alias, RealmAndHost, Filter, RT),
              Pid,
              SvcName,
              App).

%% pick_peer/5

pick_peer([], [], _, _, _) ->
    false;

%% App state is mutable but we're not in the service process: go there.
pick_peer(Local, Remote, Pid, _SvcName, #diameter_app{mutable = true} = App)
  when self() /= Pid ->
    case call_service(Pid, {pick_peer, Local, Remote, App}) of
        {TPid, _} = T when is_pid(TPid) ->
            T;
        false = No ->
            No;
        {error, _} ->
            false
    end;

%% App state isn't mutable or it is and we're in the service process:
%% do the deed.
pick_peer(Local,
          Remote,
          _Pid,
          SvcName,
          #diameter_app{alias = Alias,
                        init_state = S,
                        mutable = M}
          = App) ->
    Args = [Local, Remote, SvcName],

    try state_cb(App, pick_peer, Args) of
        {ok, {TPid, #diameter_caps{}} = T} when is_pid(TPid) ->
            T;
        {{TPid, #diameter_caps{}} = T, ModS} when is_pid(TPid), M ->
            mod_state(Alias, ModS),
            T;
        {false = No, ModS} when M ->
            mod_state(Alias, ModS),
            No;
        {ok, false = No} ->
            No;
        false = No ->
            No;
        {{TPid, #diameter_caps{}} = T, S} when is_pid(TPid) ->
            T;                     %% Accept returned state in the immutable
        {false = No, S} ->         %% case as long it isn't changed.
            No;
        T when M ->
            ModX = App#diameter_app.module,
            ?LOG(invalid_return, {ModX, pick_peer, T}),
            false
    catch
        E: Reason when M ->
            ModX = App#diameter_app.module,
            Stack = diameter_lib:get_stacktrace(),
            ?LOG(failure, {E, Reason, ModX, pick_peer, Stack}),
            false
    end.

%% peers/4

%% No peer options pointing at specific peers: search for them.
peers(Alias, RH, [Filter], T) ->
    filter(Alias, RH, Filter, T, true);

%% Or just lookup.
peers(_Alias, RH, [Filter | TPids], {PeerT, _AppT, _IdentT}) ->
    {Ts, _} = filter(caps(PeerT, TPids), RH, Filter),
    Ts.

%% filter/5
%%
%% Try to limit the peers list by starting with a host/realm lookup.

filter(Alias, RH, {neg, F}, T, B) ->
    filter(Alias, RH, F, T, not B);

filter(_, _, none, _, false) ->
    [];

filter(Alias, _, none, T, true) ->
    all_peers(Alias, T);

filter(Alias, [DR,DH] = RH, K, T, B)
  when K == realm, DR == undefined;
       K == host, DH == undefined ->
    filter(Alias, RH, none, T, B);

filter(Alias, [DR,_] = RH, realm = K, T, B) ->
    filter(Alias, RH, {K, DR}, T, B);

filter(Alias, [_,DH] = RH, host = K, T, B) ->
    filter(Alias, RH, {K, DH}, T, B);

filter(Alias, _, {K, D}, {PeerT, _AppT, IdentT}, true)
  when K == host;
       K == realm ->
    try iolist_to_binary(D) of
        B ->
            caps(PeerT, ets:select(IdentT, [{{{K, B}, '$1', '$2'},
                                             [{'==', '$1', {const, Alias}}],
                                             ['$2']}]))
    catch
        error:_ ->
            []
    end;

filter(Alias, RH, {all, Filters}, T, B)
  when is_list(Filters) ->
    fltr_all(Alias, RH, Filters, T, B);

filter(Alias, RH, {first, Filters}, T, B)
  when is_list(Filters) ->
    fltr_first(Alias, RH, Filters, T, B);

filter(Alias, RH, Filter, T, B) ->
    {Ts, Fs} = filter(all_peers(Alias, T), RH, Filter),
    choose(B, Ts, Fs).

%% fltr_all/5

fltr_all(Alias, RH, [{K, any} | Filters], T, B)
  when K == host;
       K == realm ->
    fltr_all(Alias, RH, Filters, T, B);

fltr_all(Alias, RH, [{host, _} = H, {realm, _} = R | Filters], T, B) ->
    fltr_all(Alias, RH, [R, H | Filters], T, B);

fltr_all(Alias, RH, [{realm, _} = R, {host, any} | Filters], T, B) ->
    fltr_all(Alias, RH, [R | Filters], T, B);

fltr_all(Alias, RH, [{realm, OR}, {host, OH} | Filters], T, true) ->
    {PeerT, _AppT, IdentT} = T,
    try {iolist_to_binary(OR), iolist_to_binary(OH)} of
        BT ->
            Peers = caps(PeerT,
                         ets:select(IdentT, [{{BT, '$1', '$2'},
                                              [{'==', '$1', {const, Alias}}],
                                              ['$2']}])),
            {Ts, _} = filter(Peers, RH, {all, Filters}),
            Ts
    catch
        error:_ ->
            []
    end;

fltr_all(Alias, [undefined,_] = RH, [realm | Filters], T, B) ->
    fltr_all(Alias, RH, Filters, T, B);

fltr_all(Alias, [DR,_] = RH, [realm | Filters], T, B) ->
    fltr_all(Alias, RH, [{realm, DR} | Filters], T, B);

fltr_all(Alias, [_,undefined] = RH, [host | Filters], T, B) ->
    fltr_all(Alias, RH, Filters, T, B);

fltr_all(Alias, [_,DH] = RH, [host | Filters], T, B) ->
    fltr_all(Alias, RH, [{host, DH} | Filters], T, B);

fltr_all(Alias, RH, [{K, _} = KT, KA | Filters], T, B)
  when K == host, KA == realm;
       K == realm, KA == host ->
    fltr_all(Alias, RH, [KA, KT | Filters], T, B);

fltr_all(Alias, RH, [F | Filters], T, B) ->
    {Ts, Fs} = filter(filter(Alias, RH, F, T, B), RH, {all, Filters}),
    choose(B, Ts, Fs);

fltr_all(Alias, RH, [], T, B) ->
    filter(Alias, RH, none, T, B).

%% fltr_first/5
%%
%% Like any, but stop at the first filter with any matches.

fltr_first(Alias, RH, [F | Filters], T, B) ->
    case filter(Alias, RH, F, T, B) of
        [] ->
            fltr_first(Alias, RH, Filters, T, B);
        [_|_] = Ts ->
            Ts
    end;

fltr_first(Alias, RH, [], T, B) ->
    filter(Alias, RH, none, T, not B).

%% all_peers/2

all_peers(Alias, {PeerT, AppT, _}) ->
    ets:select(PeerT, [{#peer{pid = P, caps = '$1', _ = '_'},
                        [],
                        [{{P, '$1'}}]}
                       || {_,P} <- ets:lookup(AppT, Alias)]).

%% caps/2

caps(PeerT, Pids) ->
    ets:select(PeerT, [{#peer{pid = P, caps = '$1', _ = '_'},
                        [],
                        [{{P, '$1'}}]}
                       || P <- Pids]).

%% filter/3
%%
%% Return peers in match order.

filter(Peers, _, none) ->
    {Peers, []};

filter(Peers, RH, {neg, F}) ->
    {Ts, Fs} = filter(Peers, RH, F),
    {Fs, Ts};

filter(Peers, RH, {all, L})
  when is_list(L) ->
    lists:foldl(fun(F,A) -> fltr_all(F, A, RH) end,
                {Peers, []},
                L);

filter(Peers, RH, {any, L})
  when is_list(L) ->
    lists:foldl(fun(F,A) -> fltr_any(F, A, RH) end,
                {[], Peers},
                L);

filter(Peers, RH, {first, L})
  when is_list(L) ->
    fltr_first(Peers, RH, L);

filter(Peers, RH, F) ->
    lists:partition(fun({_,C}) -> caps_filter(C, RH, F) end, Peers).

%% fltr_all/3

fltr_all(F, {Ts0, Fs0}, RH) ->
    {Ts1, Fs1} = filter(Ts0, RH, F),
    {Ts1, Fs0 ++ Fs1}.

%% fltr_any/3

fltr_any(F, {Ts0, Fs0}, RH) ->
    {Ts1, Fs1} = filter(Fs0, RH, F),
    {Ts0 ++ Ts1, Fs1}.

%% fltr_first/3

fltr_first(Peers, _, []) ->
    {[], Peers};

fltr_first(Peers, RH, [F | Filters]) ->
    case filter(Peers, RH, F) of
        {[], _} ->
            fltr_first(Peers, RH, Filters);
        {_, _} = T ->
            T
    end.

%% caps_filter/3

caps_filter(#diameter_caps{origin_host = {_,OH}}, [_,DH], host) ->
    eq(undefined, DH, OH);

caps_filter(#diameter_caps{origin_realm = {_,OR}}, [DR,_], realm) ->
    eq(undefined, DR, OR);

caps_filter(C, _, Filter) ->
    caps_filter(C, Filter).

%% caps_filter/2

caps_filter(#diameter_caps{origin_host = {_,OH}}, {host, H}) ->
    eq(any, H, OH);

caps_filter(#diameter_caps{origin_realm = {_,OR}}, {realm, R}) ->
    eq(any, R, OR);

%% Anything else is expected to be an eval filter. Filter failure is
%% documented as being equivalent to a non-matching filter.

caps_filter(C, T) ->
    try
        {eval, F} = T,
        diameter_lib:eval([F,C])
    catch
        _:_ -> false
    end.

eq(Any, Id, PeerId) ->
    Any == Id orelse try
                         iolist_to_binary(Id) == iolist_to_binary(PeerId)
                     catch
                         _:_ -> false
                     end.
%% OctetString() can be specified as an iolist() so test for string
%% rather then term equality.

%% ---------------------------------------------------------------------------
%% # service_info/2
%% ---------------------------------------------------------------------------

%% The config passed to diameter:start_service/2.
-define(CAP_INFO, ['Origin-Host',
                   'Origin-Realm',
                   'Vendor-Id',
                   'Product-Name',
                   'Origin-State-Id',
                   'Host-IP-Address',
                   'Supported-Vendor-Id',
                   'Auth-Application-Id',
                   'Inband-Security-Id',
                   'Acct-Application-Id',
                   'Vendor-Specific-Application-Id',
                   'Firmware-Revision']).

%% The config returned by diameter:service_info(SvcName, all).
-define(ALL_INFO, [capabilities,
                   applications,
                   transport,
                   options]).

%% The rest.
-define(OTHER_INFO, [connections,
                     name,
                     peers,
                     statistics,
                     info]).

service_info(Item, S)
  when is_atom(Item) ->
    case tagged_info(Item, S) of
        {_, T} -> T;
        undefined = No -> No
    end;

service_info(Items, S) ->
    tagged_info(Items, S).

tagged_info(Item, S)
  when is_atom(Item) ->
    case complete(Item) of
        {value, I} ->
            {I, complete_info(I,S)};
        false ->
            undefined
    end;

tagged_info(TPid, #state{watchdogT = WatchdogT, local = {PeerT, _, _}})
  when is_pid(TPid) ->
    try
        [#peer{watchdog = Pid}] = ets:lookup(PeerT, TPid),
        [#watchdog{ref = Ref, type = Type, options = Opts}]
            = ets:lookup(WatchdogT, Pid),
        [{ref, Ref},
         {type, Type},
         {options, Opts}]
    catch
        error:_ ->
            []
    end;

tagged_info(Items, S)
  when is_list(Items) ->
    [T || I <- Items, T <- [tagged_info(I,S)], T /= undefined, T /= []];

tagged_info(_, _) ->
    undefined.

complete_info(Item, #state{service = Svc} = S) ->
    case Item of
        name ->
            S#state.service_name;
        'Origin-Host' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.origin_host;
        'Origin-Realm' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.origin_realm;
        'Vendor-Id' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.vendor_id;
        'Product-Name' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.product_name;
        'Origin-State-Id' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.origin_state_id;
        'Host-IP-Address' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.host_ip_address;
        'Supported-Vendor-Id' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.supported_vendor_id;
        'Auth-Application-Id' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.auth_application_id;
        'Inband-Security-Id'  ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.inband_security_id;
        'Acct-Application-Id' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.acct_application_id;
        'Vendor-Specific-Application-Id' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.vendor_specific_application_id;
        'Firmware-Revision' ->
            (Svc#diameter_service.capabilities)
                #diameter_caps.firmware_revision;
        capabilities -> service_info(?CAP_INFO, S);
        applications -> info_apps(S);
        transport    -> info_transport(S);
        options      -> info_options(S);
        keys         -> ?ALL_INFO ++ ?CAP_INFO ++ ?OTHER_INFO;
        all          -> service_info(?ALL_INFO, S);
        statistics   -> info_stats(S);
        info         -> info_info(S);
        connections  -> info_connections(S);
        peers        -> info_peers(S)
    end.

complete(I)
  when I == keys;
       I == all ->
    {value, I};
complete(Pre) ->
    P = atom_to_list(Pre),
    case [I || I <- ?ALL_INFO ++ ?CAP_INFO ++ ?OTHER_INFO,
               lists:prefix(P, atom_to_list(I))]
    of
        [I] -> {value, I};
        _   -> false
    end.

%% info_stats/1

info_stats(#state{watchdogT = WatchdogT}) ->
    MatchSpec = [{#watchdog{ref = '$1', peer = '$2', _ = '_'},
                  [{'is_pid', '$2'}],
                  [['$1', '$2']]}],
    try ets:select(WatchdogT, MatchSpec) of
        L ->
            diameter_stats:read(lists:append(L))
    catch
        error: badarg -> []  %% service  has gone down
    end.

%% info_transport/1
%%
%% One entry per configured transport. Statistics for each entry are
%% the accumulated values for the ref and associated watchdog/peer
%% pids.

info_transport(S) ->
    PeerD = peer_dict(S, config_dict(S)),
    Stats = diameter_stats:sum(dict:fetch_keys(PeerD)),
    dict:fold(fun(R, Ls, A) ->
                      Cs = proplists:get_value(R, Stats, []),
                      [[{ref, R} | transport(Ls)] ++ [{statistics, Cs}] | A]
              end,
              [],
              PeerD).

%% Single config entry. Distinguish between pool_size config or not on
%% a connecting transport for backwards compatibility: with the option
%% the form is similar to the listening case, with connections grouped
%% in a pool tuple (for lack of a better name), without as before.
transport([[{type, Type}, {options, Opts}] = L])
  when Type == listen;
       Type == connect ->
    L ++ [{K, []} || [{_,K}] <- [keys(Type, Opts)]];

%% Peer entries: discard config. Note that the peer entries have
%% length at least 3.
transport([[_,_] | L]) ->
    transport(L);

%% Multiple tranports. Note that all have the same options by
%% construction, which is not terribly space efficient.
transport([[{type, Type}, {options, Opts} | _] | _] = Ls) ->
    transport(keys(Type, Opts), Ls).

%% Group transports in an accept or pool tuple ...
transport([{Type, Key}], [[{type, _}, {options, Opts} | _] | _] = Ls) ->
    [{type, Type},
     {options, Opts},
     {Key, [tl(tl(L)) || L <- Ls]}];

%% ... or not: there can only be one.
transport([], [L]) ->
    L.

keys(connect = T, Opts) ->
    [{T, pool} || lists:keymember(pool_size, 1, Opts)];
keys(_, _) ->
    [{listen, accept}].

peer_dict(#state{watchdogT = WatchdogT, local = {PeerT, _, _}}, Dict0) ->
    try ets:tab2list(WatchdogT) of
        L -> lists:foldl(fun(T,A) -> peer_acc(PeerT, A, T) end, Dict0, L)
    catch
        error: badarg -> Dict0  %% service has gone down
    end.

peer_acc(PeerT, Acc, #watchdog{pid = Pid,
                               type = Type,
                               ref = Ref,
                               options = Opts,
                               state = WS,
                               started = At,
                               peer = TPid}) ->
    Info = [{type, Type},
            {options, Opts},
            {watchdog, {Pid, At, WS}}
            | info_peer(PeerT, TPid, WS)],
    dict:append(Ref, Info ++ [{info, info_process_info(Info)}], Acc).

info_peer(PeerT, TPid, WS)
  when is_pid(TPid), WS /= ?WD_DOWN ->
    try ets:lookup(PeerT, TPid) of
        T -> info_peer(T)
    catch
        error: badarg -> []  %% service has gone down
    end;
info_peer(_, _, _) ->
    [].

info_process_info(Info) ->
    lists:flatmap(fun ipi/1, Info).

ipi({watchdog, {Pid, _, _}}) ->
    info_pid(Pid);

ipi({peer, {Pid, _}}) ->
    info_pid(Pid);

ipi({port, [{owner, Pid} | _]}) ->
    info_pid(Pid);

ipi(_) ->
    [].

info_pid(Pid) ->
    case process_info(Pid, [message_queue_len, memory, binary]) of
        undefined ->
            [];
        L ->
            [{Pid, lists:map(fun({K,V}) -> {K, map_info(K,V)} end, L)}]
    end.

%% The binary list consists of 3-tuples {Ptr, Size, Count}, where Ptr
%% is a C pointer value, Size is the size of a referenced binary in
%% bytes, and Count is a global reference count. The same Ptr can
%% occur multiple times, once for each reference on the process heap.
%% In this case, the corresponding tuples will have Size in common but
%% Count may differ just because no global lock is taken when the
%% value is retrieved.
%%
%% The list can be quite large, and we aren't often interested in the
%% pointers or counts, so whittle this down to the number of binaries
%% referenced and their total byte count.
map_info(binary, L) ->
    SzD = lists:foldl(fun({P,S,_}, D) -> dict:store(P,S,D) end,
                      dict:new(),
                      L),
    {dict:size(SzD), dict:fold(fun(_,S,N) -> S + N end, 0, SzD)};

map_info(_, T) ->
    T.

%% The point of extracting the config here is so that 'transport' info
%% has one entry for each transport ref, the peer table only
%% containing entries that have a living watchdog.

config_dict(#state{service_name = SvcName}) ->
    lists:foldl(fun config_acc/2,
                dict:new(),
                diameter_config:lookup(SvcName)).

config_acc({Ref, T, Opts}, Dict)
  when T == listen;
       T == connect ->
    dict:store(Ref, [[{type, T}, {options, Opts}]], Dict);
config_acc(_, Dict) ->
    Dict.

info_peer([#peer{pid = Pid, apps = SApps, caps = Caps, started = T}]) ->
    [{peer, {Pid, T}},
     {apps, SApps},
     {caps, info_caps(Caps)}
     | try [{port, info_port(Pid)}] catch _:_ -> [] end];
info_peer([] = No) ->
    No.

%% Extract information that the processes involved are expected to
%% "publish" in their process dictionaries. Simple but backhanded.
info_port(Pid) ->
    {_, PD} = process_info(Pid, dictionary),
    {_, T} = lists:keyfind({diameter_peer_fsm, start}, 1, PD),
    {TPid, {_Type, TMod, _Cfg}} = T,
    {_, TD} = process_info(TPid, dictionary),
    {_, Data} = lists:keyfind({TMod, info}, 1, TD),
    [{owner, TPid},
     {module, TMod}
     | try TMod:info(Data) catch _:_ -> [] end].

%% Use the fields names from diameter_caps instead of
%% diameter_base_CER to distinguish between the 2-tuple values
%% compared to the single capabilities values. Note also that the
%% returned list is tagged 'caps' rather than 'capabilities' to
%% emphasize the difference.
info_caps(#diameter_caps{} = C) ->
    lists:zip(record_info(fields, diameter_caps), tl(tuple_to_list(C))).

info_apps(#state{service = #diameter_service{applications = Apps}}) ->
    lists:map(fun mk_app/1, Apps).

mk_app(#diameter_app{} = A) ->
    lists:zip(record_info(fields, diameter_app), tl(tuple_to_list(A))).

%% info_info/1
%%
%% Extract process_info from connections info.

info_info(S) ->
    [I || L <- conn_list(S), {info, I} <- L].

%% info_connections/1
%%
%% One entry per transport connection. Statistics for each entry are
%% for the peer pid only.

info_connections(S) ->
    ConnL = conn_list(S),
    Stats = diameter_stats:read([P || L <- ConnL, {peer, {P,_}} <- L]),
    [L ++ [stats([P], Stats)] || L <- ConnL, {peer, {P,_}} <- L].

conn_list(S) ->
    lists:append(dict:fold(fun conn_acc/3, [], peer_dict(S, dict:new()))).

conn_acc(Ref, Peers, Acc) ->
    [[[{ref, Ref} | L] || L <- Peers, lists:keymember(peer, 1, L)]
     | Acc].

stats(Refs, Stats) ->
    {statistics, dict:to_list(lists:foldl(fun(R,D) ->
                                                  stats_acc(R, D, Stats)
                                          end,
                                          dict:new(),
                                          Refs))}.

stats_acc(Ref, Dict, Stats) ->
    lists:foldl(fun({C,N}, D) -> dict:update_counter(C, N, D) end,
                Dict,
                proplists:get_value(Ref, Stats, [])).

%% info_peers/1
%%
%% One entry per peer Origin-Host. Statistics for each entry are
%% accumulated values for all peer pids.

info_peers(S) ->
    {PeerD, RefD} = lists:foldl(fun peer_acc/2,
                                {dict:new(), dict:new()},
                                conn_list(S)),
    Refs = lists:append(dict:fold(fun(_, Rs, A) -> [Rs|A] end,
                                  [],
                                  RefD)),
    Stats = diameter_stats:read(Refs),
    dict:fold(fun(OH, Cs, A) ->
                      Rs = dict:fetch(OH, RefD),
                      [{OH, [{connections, Cs}, stats(Rs, Stats)]} | A]
              end,
              [],
              PeerD).

peer_acc(Peer, {PeerD, RefD}) ->
    [{TPid, _}, [{origin_host, {_, OH}} | _]]
        = [proplists:get_value(K, Peer) || K <- [peer, caps]],
    {dict:append(OH, Peer, PeerD), dict:append(OH, TPid, RefD)}.

%% info_options/1

info_options(S) ->
    S#state.options.