Send events for connection establishment failure

If a peer fsm process exits then the exit reason is received by
the service process in a 'DOWN' message. If the reason is the one
generated by diameter_peer_fsm:close/2, which is called to signal
a non-transport failure before the completion of capabilities exchange
(eg. receiving an unsuccessful CEA), then an event is sent to any
subscribers.

Also, tweak capabilities_cb return values for more informative
event data.

4 files changed, 144 insertions, 142 deletions

diff --git a/lib/diameter/src/base/diameter_capx.erl b/lib/diameter/src/base/diameter_capx.erl
index 293eb0c196..842a9e6103 100644
--- a/lib/diameter/src/base/diameter_capx.erl
+++ b/lib/diameter/src/base/diameter_capx.erl
@@ -57,9 +57,9 @@
 -include("diameter_types.hrl").
 -include("diameter_gen_base_rfc3588.hrl").
 
--define(SUCCESS, ?'DIAMETER_BASE_RESULT-CODE_DIAMETER_SUCCESS').
--define(NOAPP, ?'DIAMETER_BASE_RESULT-CODE_DIAMETER_NO_COMMON_APPLICATION').
--define(NOSECURITY, ?'DIAMETER_BASE_RESULT-CODE_DIAMETER_NO_COMMON_SECURITY').
+-define(SUCCESS,    2001).  %% DIAMETER_SUCCESS
+-define(NOAPP,      5010).  %% DIAMETER_NO_COMMON_APPLICATION
+-define(NOSECURITY, 5017).  %% DIAMETER_NO_COMMON_SECURITY
 
 -define(NO_INBAND_SECURITY, 0).
 -define(TLS, 1).
@@ -96,7 +96,7 @@ try_it([Fun | Args]) ->
     try apply(Fun, Args) of
         T -> {ok, T}
     catch
-        throw: ?FAILURE(Reason) -> {error, {Reason, Args}}
+        throw: ?FAILURE(Reason) -> {error, Reason}
     end.
 
 %% mk_caps/2
@@ -288,25 +288,12 @@ to_cea(CER, Field, CEA) ->
         
 %% rCEA/2
 
-rCEA(#diameter_base_CEA{'Result-Code' = RC}
-     = CEA,
-     #diameter_service{capabilities = LCaps}
-     = Svc) ->
-    RC == ?SUCCESS orelse ?THROW({'Result-Code', RC}),
-
+rCEA(CEA, #diameter_service{capabilities = LCaps} = Svc) ->
     RCaps = capx_to_caps(CEA),
     SApps = common_applications(LCaps, RCaps, Svc),
-
-    [] == SApps andalso ?THROW(no_common_applications),
-
     IS = common_security(LCaps, RCaps),
 
-    [] == IS andalso ?THROW(no_common_security),
-
-    {SApps, IS, RCaps};
-
-rCEA(CEA, _Svc) ->
-    ?THROW({invalid, CEA}).
+    {SApps, IS, RCaps}.
 
 %% capx_to_caps/1
 
diff --git a/lib/diameter/src/base/diameter_codec.erl b/lib/diameter/src/base/diameter_codec.erl
index d88f42fb7c..fe1212b7e0 100644
--- a/lib/diameter/src/base/diameter_codec.erl
+++ b/lib/diameter/src/base/diameter_codec.erl
@@ -190,26 +190,13 @@ encode_avps(Avps) ->
 
 %% msg_header/3
 
-msg_header(Mod, MsgName, Header) ->
-    {Code, Flags, ApplId} = h(Mod, MsgName, Header),
-    {Code, p(Flags, Header), ApplId}.
-
-%% 6.2 of 3588 requires the same 'P' bit on an answer as on the
-%% request.
-
-p(Flags, #diameter_header{is_request = true,
-                          is_proxiable = P}) ->
-    Flags band (2#10110000 bor choose(P, 2#01000000, 0));
-p(Flags, _) ->
-    Flags.
-
-h(Mod, 'answer-message' = MsgName, Header) ->
+msg_header(Mod, 'answer-message' = MsgName, Header) ->
     ?BASE = Mod,
     #diameter_header{cmd_code = Code} = Header,
     {_, Flags, ApplId} = ?BASE:msg_header(MsgName),
     {Code, Flags, ApplId};
 
-h(Mod, MsgName, _) ->
+msg_header(Mod, MsgName, _) ->
     Mod:msg_header(MsgName).
 
 %% rec2msg/2
@@ -554,8 +541,3 @@ pack_avp(Code, Flags, Vid, Sz, Bin) ->
 pack_avp(Code, Flags, Sz, Bin) ->
     Length = Sz + 8,
     <<Code:32, Flags:8, Length:24, Bin/binary>>.
-
-%% ===========================================================================
-
-choose(true, X, _)  -> X;
-choose(false, _, X) -> X.
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index 3f1610b325..d754c1bcc8 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -325,9 +325,10 @@ send_CER(#state{mode = {connect, Remote},
                 service = #diameter_service{capabilities = Caps},
                 transport = TPid}
          = S) ->
-    req_send_CER(Caps#diameter_caps.origin_host, Remote)
+    OH = Caps#diameter_caps.origin_host,
+    req_send_CER(OH, Remote)
         orelse
-        close(connected, S),
+        close({already_connected, Remote, Caps}, S),
     CER = build_CER(S),
     ?LOG(send, 'CER'),
     send(TPid, encode(CER)),
@@ -469,19 +470,19 @@ handle_request(Type, #diameter_packet{} = Pkt, S) ->
 %% send_answer/3
 
 send_answer(Type, ReqPkt, #state{transport = TPid} = S) ->
-    #diameter_packet{header = #diameter_header{version = V,
-                                               end_to_end_id = Eid,
-                                               hop_by_hop_id = Hid,
-                                               is_proxiable = P},
+    #diameter_packet{header = H,
                      transport_data = TD}
         = ReqPkt,
 
-    {Msg, PostF} = build_answer(Type, V, ReqPkt, S),
+    {Msg, PostF} = build_answer(Type, ReqPkt, S),
 
-    Pkt = #diameter_packet{header = #diameter_header{version = V,
-                                                     end_to_end_id = Eid,
-                                                     hop_by_hop_id = Hid,
-                                                     is_proxiable = P},
+    %% An answer message clears the R and T flags and retains the P
+    %% flag. The E flag is set at encode.
+    Pkt = #diameter_packet{header
+                           = H#diameter_header{version = ?DIAMETER_VERSION,
+                                               is_request = false,
+                                               is_error = undefined,
+                                               is_retransmitted = false},
                            msg = Msg,
                            transport_data = TD},
 
@@ -493,57 +494,79 @@ eval([F|A], S) ->
 eval(ok, S) ->
     S.
 
-%% build_answer/4
+%% build_answer/3
 
 build_answer('CER',
-             ?DIAMETER_VERSION,
              #diameter_packet{msg = CER,
-                              header = #diameter_header{is_error = false},
+                              header = #diameter_header{version
+                                                        = ?DIAMETER_VERSION,
+                                                        is_error = false},
                               errors = []}
              = Pkt,
-             #state{service = Svc}
-             = S) ->
+             S) ->
     {SupportedApps, RCaps, #diameter_base_CEA{'Result-Code' = RC,
-                                              'Inband-Security-Id' = [IS]}
+                                              'Inband-Security-Id' = IS}
                            = CEA}
         = recv_CER(CER, S),
 
-    #diameter_service{capabilities = LCaps}
-        = Svc,
-
     #diameter_caps{origin_host = {OH, DH}}
         = Caps
-        = capz(LCaps, RCaps),
+        = capz(caps(S), RCaps),
 
     try
         2001 == RC  %% DIAMETER_SUCCESS
-            orelse ?THROW({result_code, RC}),
+            orelse ?THROW(RC),
         register_everywhere({?MODULE, connection, OH, DH})
-            orelse ?THROW({result_code, 4003}),  %% DIAMETER_ELECTION_LOST
+            orelse ?THROW(4003),  %% DIAMETER_ELECTION_LOST
         caps_cb(Caps)
     of
-        ok -> {CEA, [fun open/5, Pkt, SupportedApps, Caps, {accept, IS}]}
+        N -> {cea(CEA, N), [fun open/5, Pkt,
+                                        SupportedApps,
+                                        Caps,
+                                        {accept, hd([_] = IS)}]}
     catch
-        ?FAILURE(discard = T) ->
-            close({'CER', T, DH}, S);
-        ?FAILURE({result_code, N}) ->
-            {answer_message(cea(S), N), [fun close/2, {'CER', N, DH}]}
+        ?FAILURE(Reason) ->
+            rejected(Reason, {'CER', Reason, Caps, Pkt}, S)
     end;
 
 %% The error checks below are similar to those in diameter_service for
 %% other messages. Should factor out the commonality.
 
-build_answer(Type, V, #diameter_packet{header = H, errors = Es} = Pkt, S) ->
-    FailedAvp = failed_avp([A || {_,A} <- Es]),
-    Msg = answer_message(answer(Type, S), rc(V, H, Es)),
-    {set(Msg, FailedAvp), if 'CER' == Type ->
-                                  [fun close/2, {Type, V, Pkt}];
-                             true ->
-                                  ok
-                          end}.
+build_answer(Type,
+             #diameter_packet{header = H,
+                              errors = Es}
+             = Pkt,
+             S) ->
+    RC = rc(H, Es),
+    {answer(Type, RC, Es, S), post(Type, RC, Pkt, S)}.
+
+cea(CEA, ok) ->
+    CEA;
+cea(CEA, 2001) ->
+    CEA;
+cea(CEA, RC) ->
+    CEA#diameter_base_CEA{'Result-Code' = RC}.
+
+post('CER' = T, RC, Pkt, S) ->
+    [fun close/2, {T, caps(S), {RC, Pkt}}];
+post(_, _, _, _) ->
+    ok.
+
+rejected({capabilities_cb, _F, Reason}, T, S) ->
+    rejected(Reason, T, S);
 
-cea(S) ->
-    answer('CER', S).
+rejected(discard, T, S) ->
+    close(T, S);
+rejected({N, Es}, T, S) ->
+    {answer('CER', N, Es, S), [fun close/2, T]};
+rejected(N, T, S) ->
+    rejected({N, []}, T, S).
+
+answer(Type, RC, Es, S) ->
+    set(answer(Type, RC, S), failed_avp([A || {_,A} <- Es])).
+
+answer(Type, RC, S) ->
+    answer_message(answer(Type, S), RC).
 
 %% answer_message/2
 
@@ -576,19 +599,19 @@ set(['answer-message' | _] = Ans, FailedAvp) ->
 set([_|_] = Ans, FailedAvp) ->
     Ans ++ FailedAvp.
 
-%% rc/3
+%% rc/2
 
-rc(_, #diameter_header{is_error = true}, _) ->
+rc(#diameter_header{is_error = true}, _) ->
     3008;  %% DIAMETER_INVALID_HDR_BITS
 
-rc(_, _, [Bs|_])
+rc(_, [Bs|_])
   when is_bitstring(Bs) ->
     3009;  %% DIAMETER_INVALID_HDR_BITS
 
-rc(?DIAMETER_VERSION, _, Es) ->
+rc(#diameter_header{version = ?DIAMETER_VERSION}, Es) ->
     rc(Es);
 
-rc(_, _, _) ->
+rc(_, _) ->
     5011.  %% DIAMETER_UNSUPPORTED_VERSION
 
 %% rc/1
@@ -656,58 +679,68 @@ recv_CER(CER, #state{service = Svc}) ->
 
 %% handle_CEA/1
 
-handle_CEA(#diameter_packet{header = #diameter_header{version = V},
-                            bin = Bin}
+handle_CEA(#diameter_packet{bin = Bin}
            = Pkt,
            #state{service = #diameter_service{capabilities = LCaps}}
            = S)
   when is_binary(Bin) ->
     ?LOG(recv, 'CEA'),
 
-    ?DIAMETER_VERSION == V orelse close({version, V}, S),
-
-    #diameter_packet{msg = CEA, errors = Errors}
+    #diameter_packet{msg = CEA}
         = DPkt
         = diameter_codec:decode(?BASE, Pkt),
 
-    [] == Errors orelse close({errors, Errors}, S),
-
-    {SApps, [IS], RCaps} = recv_CEA(CEA, S),
+    {SApps, IS, RCaps} = recv_CEA(DPkt, S),
 
     #diameter_caps{origin_host = {OH, DH}}
         = Caps
         = capz(LCaps, RCaps),
 
+    #diameter_base_CEA{'Result-Code' = RC}
+        = CEA,
+
     %% Ensure that we don't already have a connection to the peer in
     %% question. This isn't the peer election of 3588 except in the
     %% sense that, since we don't know who we're talking to until we
     %% receive a CER/CEA, the first that arrives wins the right to a
     %% connection with the peer.
 
-    register_everywhere({?MODULE, connection, OH, DH})
-        orelse close({'CEA', DH}, S),
-
-    try caps_cb(Caps) of
-        ok -> open(DPkt, SApps, Caps, {connect, IS}, S)
+    try
+        2001 == RC
+            orelse ?THROW(RC),
+        [] == SApps
+            andalso ?THROW(no_common_application),
+        [] == IS
+            andalso ?THROW(no_common_security),
+        register_everywhere({?MODULE, connection, OH, DH})
+            orelse ?THROW(election_lost),
+        caps_cb(Caps)
+    of
+        _ -> open(DPkt, SApps, Caps, {connect, hd([_] = IS)}, S)
     catch
-        ?FAILURE(Reason) -> close(Reason, S)
+        ?FAILURE(Reason) -> close({'CEA', Reason, Caps, DPkt}, S)
     end.
+%% Check more than the result code since the peer could send 2001
+%% regardless.
 
 %% recv_CEA/2
 
-recv_CEA(CEA, #state{service = Svc} = S) ->
-    case diameter_capx:recv_CEA(CEA, Svc) of
-        {ok, {_,_}} -> %% return from old code
-            close({'CEA', update}, S);
-        {ok, {[], _, _}} ->
-            close({'CEA', no_common_application}, S);
-        {ok, {_, [], _}} ->
-            close({'CEA', no_common_security}, S);
-        {ok, {_,_,_} = T} ->
-            T;
-        {error, Reason} ->
-            close({'CEA', Reason}, S)
-    end.
+recv_CEA(#diameter_packet{header = #diameter_header{version
+                                                    = ?DIAMETER_VERSION,
+                                                    is_error = false},
+                          msg = CEA,
+                          errors = []},
+         #state{service = Svc}) ->
+    {ok, T} = diameter_capx:recv_CEA(CEA, Svc),
+    T;
+
+recv_CEA(Pkt, S) ->
+    close({'CEA', caps(S), Pkt}, S).
+
+caps(#diameter_service{capabilities = Caps}) ->
+    Caps;
+caps(#state{service = Svc}) ->
+    caps(Svc).
 
 %% caps_cb/1
 
@@ -721,17 +754,21 @@ ccb([F | Rest], T) ->
     case diameter_lib:eval([F|T]) of
         ok ->
             ccb(Rest, T);
+        N when 2 == N div 1000 ->  %% 2xxx Result-Code
+            N;
         Res ->
-            ?THROW({{capabilities_cb, F}, rejected(Res)})
+            ?THROW({capabilities_cb, F, rejected(Res)})
     end.
+%% Note that returning 2xxx causes the capabilities exchange to be
+%% accepted directly, without further callbacks.
 
-rejected({result_code, N} = T)
-  when 1000 =< N, N < 6000 ->
-    T;
 rejected(discard = T) ->
     T;
 rejected(unknown) ->
-    {result_code, 3010}.  %% DIAMETER_UNKNOWN_PEER
+    3010;  %% DIAMETER_UNKNOWN_PEER
+rejected(N)
+  when is_integer(N) ->
+    N.
 
 %% open/5
 
@@ -740,26 +777,26 @@ open(Pkt, SupportedApps, Caps, {Type, IS}, #state{parent = Pid} = S) ->
                    inband_security_id = {LS,_}}
         = Caps,
 
-    tls_ack(lists:member(?TLS, LS), Type, IS, S),
+    tls_ack(lists:member(?TLS, LS), Caps, Type, IS, S),
     Pid ! {open, self(), H, {Caps, SupportedApps, Pkt}},
 
     S#state{state = 'Open'}.
 
 %% We've advertised TLS support: tell the transport the result
 %% and expect a reply when the handshake is complete.
-tls_ack(true, Type, IS, #state{transport = TPid} = S) ->
+tls_ack(true, Caps, Type, IS, #state{transport = TPid} = S) ->
     Ref = make_ref(),
     TPid ! {diameter, {tls, Ref, Type, IS == ?TLS}},
     receive
         {diameter, {tls, Ref}} ->
             ok;
-        {'DOWN', _, process, TPid, _} = T ->
-            close({tls_ack, T}, S)
+        {'DOWN', _, process, TPid, Reason} ->
+            close({tls_ack, Reason, Caps}, S)
     end;
 
 %% Or not. Don't send anything to the transport so that transports
 %% not supporting TLS work as before without modification.
-tls_ack(false, _, _, _) ->
+tls_ack(false, _, _, _, _) ->
     ok.
 
 capz(#diameter_caps{} = L, #diameter_caps{} = R) ->
diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl
index 421e36ccf5..3ae6ed755f 100644
--- a/lib/diameter/src/base/diameter_service.erl
+++ b/lib/diameter/src/base/diameter_service.erl
@@ -983,7 +983,8 @@ peer_cb(MFA, Alias) ->
 connection_down(Pid, #state{peerT = PeerT,
                             connT = ConnT}
                      = S) ->
-    #peer{conn = TPid}
+    #peer{op_state = ?STATE_UP,  %% assert
+          conn = TPid}
         = P
         = fetch(PeerT, Pid),
 
@@ -993,6 +994,9 @@ connection_down(Pid, #state{peerT = PeerT,
 
 %% connection_down/3
 
+connection_down(#peer{op_state = ?STATE_DOWN}, _, S) ->
+    S;
+
 connection_down(#peer{conn = TPid,
                       op_state = ?STATE_UP}
                 = P,
@@ -1034,13 +1038,23 @@ down_conn(Id, Alias, TC, {SvcName, Apps}) ->
 
 %% Peer process has died.
 
-peer_down(Pid, _Reason, #state{peerT = PeerT} = S) ->
+peer_down(Pid, Reason, #state{peerT = PeerT} = S) ->
     P = fetch(PeerT, Pid),
     ets:delete_object(PeerT, P),
+    closed(Reason, P, S),
     restart(P,S),
     peer_down(P,S).
 
-%% peer_down/2
+%% Send an event at connection establishment failure.
+closed({shutdown, {close, _TPid, Reason}},
+       #peer{op_state = ?STATE_DOWN,
+             ref = Ref,
+             type = Type,
+             options = Opts},
+       #state{service_name = SvcName}) ->
+    send_event(SvcName, {closed, Ref, Reason, {type(Type), Opts}});
+closed(_, _, _) ->
+    ok.
 
 %% The peer has never come up ...
 peer_down(#peer{conn = B}, S)
@@ -1048,27 +1062,9 @@ peer_down(#peer{conn = B}, S)
     S;
 
 %% ... or it has.
-peer_down(#peer{ref = Ref,
-                conn = TPid,
-                type = Type,
-                options = Opts}
-          = P,
-          #state{service_name = SvcName,
-                 connT = ConnT}
-          = S) ->
-    #conn{caps = Caps}
-        = C
-        = fetch(ConnT, TPid),
+peer_down(#peer{conn = TPid} = P, #state{connT = ConnT} = S) ->
+    #conn{} = C = fetch(ConnT, TPid),
     ets:delete_object(ConnT, C),
-    try
-        pd(P,C,S)
-    after
-        send_event(SvcName, {closed, Ref, {TPid, Caps}, {type(Type), Opts}})
-    end.
-
-pd(#peer{op_state = ?STATE_DOWN}, _, S) ->
-    S;
-pd(#peer{op_state = ?STATE_UP} = P, C, S) ->
     connection_down(P,C,S).
 
 %% restart/2