Merge branch 'anders/diameter/dpr/OTP-12542' into maint

* anders/diameter/dpr/OTP-12542:
  Discard CER or DWR sent with diameter:call/4
  Allow DPR to be sent with diameter:call/4
  Add transport_opt() dpa_timeout
  Add testcase for sending DPR with diameter:call/4

8 files changed, 159 insertions, 43 deletions

diff --git a/lib/diameter/doc/src/diameter.xml b/lib/diameter/doc/src/diameter.xml
index 638c1c4c2b..27f2d76f28 100644
--- a/lib/diameter/doc/src/diameter.xml
+++ b/lib/diameter/doc/src/diameter.xml
@@ -1115,7 +1115,7 @@ Defaults to <c>rebooting</c> for <c>Reason=service|application</c> and
 <p>
 Number of milliseconds after which the transport process is
 terminated if DPA has not been received.
-Defaults to 1000.</p>
+Defaults to the value of &dpa_timeout;.</p>
 </item>
 </taglist>
 </item>
@@ -1152,6 +1152,17 @@ configured them.</p>
 Defaults to a single callback returning <c>dpr</c>.</p>
 </item>
 
+<marker id="dpa_timeout"/>
+<tag><c>{dpa_timeout, &dict_Unsigned32;}</c></tag>
+<item>
+<p>
+Number of milliseconds after which a transport connection is
+terminated following an outgoing DPR if DPA is not received.</p>
+
+<p>
+Defaults to 1000.</p>
+</item>
+
 <marker id="length_errors"/>
 <tag><c>{length_errors, exit|handle|discard}</c></tag>
 <item>
diff --git a/lib/diameter/doc/src/seealso.ent b/lib/diameter/doc/src/seealso.ent
index 44541afb9b..8d87a4a5e7 100644
--- a/lib/diameter/doc/src/seealso.ent
+++ b/lib/diameter/doc/src/seealso.ent
@@ -4,7 +4,7 @@
 
 %CopyrightBegin%
 
-Copyright Ericsson AB 2012-2014. All Rights Reserved.
+Copyright Ericsson AB 2012-2015. All Rights Reserved.
 
 The contents of this file are subject to the Erlang Public License,
 Version 1.1, (the "License"); you may not use this file except in
@@ -64,6 +64,7 @@ significant.
 <!ENTITY capabilities_cb '<seealso marker="#capabilities_cb">capabilities_cb</seealso>'>
 <!ENTITY capx_timeout '<seealso marker="#capx_timeout">capx_timeout</seealso>'>
 <!ENTITY disconnect_cb '<seealso marker="#disconnect_cb">disconnect_cb</seealso>'>
+<!ENTITY dpa_timeout '<seealso marker="#dpa_timeout">dpa_timeout</seealso>'>
 <!ENTITY transport_config '<seealso marker="#transport_config">transport_config</seealso>'>
 <!ENTITY transport_module '<seealso marker="#transport_module">transport_module</seealso>'>
 <!ENTITY connect_timer '<seealso marker="#connect_timer">connect_timer</seealso>'>
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl
index 1bbdf6e34d..a45d84f95b 100644
--- a/lib/diameter/src/base/diameter.erl
+++ b/lib/diameter/src/base/diameter.erl
@@ -343,6 +343,7 @@ call(SvcName, App, Message) ->
     | {capabilities_cb, evaluable()}
     | {capx_timeout, 'Unsigned32'()}
     | {disconnect_cb, evaluable()}
+    | {dpa_timeout, 'Unsigned32'()}
     | {length_errors, exit | handle | discard}
     | {connect_timer, 'Unsigned32'()}
     | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}}
diff --git a/lib/diameter/src/base/diameter_config.erl b/lib/diameter/src/base/diameter_config.erl
index c0a4f7df69..aa4d6e5a20 100644
--- a/lib/diameter/src/base/diameter_config.erl
+++ b/lib/diameter/src/base/diameter_config.erl
@@ -532,7 +532,9 @@ opt({applications, As}) ->
 opt({capabilities, Os}) ->
     is_list(Os) andalso ok == encode_CER(Os);
 
-opt({capx_timeout, Tmo}) ->
+opt({K, Tmo})
+  when K == capx_timeout;
+       K == dpa_timeout ->
     ?IS_UINT32(Tmo);
 
 opt({length_errors, T}) ->
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index 23bba701eb..9ff6845ab7 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -63,6 +63,7 @@
 %% Keys in process dictionary.
 -define(CB_KEY, cb).         %% capabilities callback
 -define(DPR_KEY, dpr).       %% disconnect callback
+-define(DPA_KEY, dpa).       %% timeout for DPA reception
 -define(REF_KEY, ref).       %% transport_ref()
 -define(Q_KEY, q).           %% transport start queue
 -define(START_KEY, start).   %% start of connected transport
@@ -107,7 +108,8 @@
          transport    :: pid(),     %% transport process
          dictionary   :: module(),  %% common dictionary
          service      :: #diameter_service{},
-         dpr = false  :: false | {uint32(), uint32()},
+         dpr = false  :: false | {uint32(), uint32()}  %% set in old code
+                               | {boolean(), uint32(), uint32()},
                             %% | hop by hop and end to end identifiers
          length_errors :: exit | handle | discard}).
 
@@ -187,6 +189,7 @@ i({Ack, WPid, {M, Ref} = T, Opts, {Mask, Nodes, Dict0, Svc}}) ->
     putr(?REF_KEY, Ref),
     putr(?SEQUENCE_KEY, Mask),
     putr(?RESTRICT_KEY, Nodes),
+    putr(?DPA_KEY, proplists:get_value(dpa_timeout, Opts, ?DPA_TIMEOUT)),
 
     Tmo = proplists:get_value(capx_timeout, Opts, ?EVENT_TIMEOUT),
     OnLengthErr = proplists:get_value(length_errors, Opts, exit),
@@ -212,9 +215,12 @@ wait(Ref, Pid) ->
         Ref ->
             ok;
         {'DOWN', _, process, Pid, _} = D ->
-            exit({shutdown, D})
+            x(D)
     end.
 
+x(T) ->
+    exit({shutdown, T}).
+
 start_transport(T, Opts, #diameter_service{capabilities = LCaps} = Svc) ->
     Addrs0 = LCaps#diameter_caps.host_ip_address,
     start_transport(Addrs0, {T, Opts, Svc}).
@@ -226,7 +232,7 @@ start_transport(Addrs0, T) ->
             q_next(TPid, Addrs0, Tmo, Data),
             {TPid, Addrs};
         {error, No} ->
-            exit({shutdown, {no_connection, No}})
+            x({no_connection, No})
     end.
 
 svc(#diameter_service{capabilities = LCaps0} = Svc, Addrs) ->
@@ -333,6 +339,9 @@ eraser(Key) ->
 
 %% transition/2
 
+transition(T, #state{dpr = {Hid, Eid}} = S) -> %% DPR sent from old code
+    transition(T, S#state{dpr = {false, Hid, Eid}});
+
 %% Connection to peer.
 transition({diameter, {TPid, connected, Remote}},
            #state{transport = TPid,
@@ -398,8 +407,7 @@ transition({timeout, _}, _) ->
 
 %% Outgoing message.
 transition({send, Msg}, S) ->
-    outgoing(Msg, S),
-    ok;
+    outgoing(Msg, S);
 
 %% Request for graceful shutdown at remove_transport, stop_service of
 %% application shutdown.
@@ -535,10 +543,23 @@ recv(Bin, S) ->
 %% DPR, so both ends don't do so when sending simultaneously.
 recv1(Name,
       #diameter_packet{header = #diameter_header{is_request = true} = H},
-      #state{dpr = {_,_}})
+      #state{dpr = {_,_,_}})
   when Name /= 'DPR' ->
     invalid(false, recv_after_dpr, H);
 
+%% DPA with identifier mismatch, or in response to a DPR initiated by
+%% the service.
+recv1('DPA' = N,
+      #diameter_packet{header = #diameter_header{hop_by_hop_id = Hid,
+                                                 end_to_end_id = Eid}}
+      = Pkt,
+      #state{dpr = {X,H,E}}
+      = S)
+  when H /= Hid;
+       E /= Eid;
+       not X ->
+    rcv(N, Pkt, S);
+
 %% Any other message with a header and no length errors: send to the
 %% parent.
 recv1(Name, Pkt, #state{parent = Pid} = S) ->
@@ -601,7 +622,7 @@ rcv(Name, _, #state{state = PS})
 rcv('DPR' = N, Pkt, S) ->
     handle_request(N, Pkt, S);
 
-%% DPA in response to DPR and with the expected identifiers.
+%% DPA in response to DPR, with the expected identifiers.
 rcv('DPA' = N,
     #diameter_packet{header = #diameter_header{end_to_end_id = Eid,
                                                hop_by_hop_id = Hid}
@@ -609,10 +630,15 @@ rcv('DPA' = N,
     = Pkt,
     #state{dictionary = Dict0,
            transport = TPid,
-           dpr = {Hid, Eid}}) ->
+           dpr = {X, Hid, Eid}}) ->
     ?LOG(recv, N),
-    incr(recv, H, Dict0),
-    incr_rc(recv, diameter_codec:decode(Dict0, Pkt), Dict0),
+    X orelse begin
+                 %% Only count DPA in response to a DPR sent by the
+                 %% service: explicit DPR is counted in the same way
+                 %% as other explicitly sent requests.
+                 incr(recv, H, Dict0),
+                 incr_rc(recv, diameter_codec:decode(Dict0, Pkt), Dict0)
+             end,
     diameter_peer:close(TPid),
     {stop, N};
 
@@ -650,23 +676,52 @@ incr_error(Dir, Pkt, Dict0) ->
 %% Msg here could be a #diameter_packet or a binary depending on who's
 %% sending. In particular, the watchdog will send DWR as a binary
 %% while messages coming from clients will be in a #diameter_packet.
+
 send(Pid, Msg) ->
     diameter_peer:send(Pid, Msg).
 
 %% outgoing/2
 
+%% Explicit DPR.
+outgoing(#diameter_packet{header = #diameter_header{application_id = 0,
+                                                    cmd_code = 282,
+                                                    is_request = true}
+                                 = H}
+         = Pkt,
+         #state{dpr = T,
+                parent = Pid}
+         = S) ->
+    if T == false ->
+            inform_dpr(Pid),
+            send_dpr(true, Pkt, dpa_timeout(), S);
+       true ->
+            invalid(false, dpr_after_dpr, H)  %% already sent: discard
+    end;
+
+%% Explict CER or DWR: discard. These are sent by us.
+outgoing(#diameter_packet{header = #diameter_header{application_id = 0,
+                                                    cmd_code = C,
+                                                    is_request = true}
+                                 = H},
+         _)
+  when 257 == C;    %% CER
+       280 == C ->  %% DWR
+    invalid(false, invalid_request, H);
+
 %% DPR not sent: send.
 outgoing(Msg, #state{transport = TPid, dpr = false}) ->
-    send(TPid, Msg);
+    send(TPid, Msg),
+    ok;
 
 %% Outgoing answer: send.
 outgoing(#diameter_packet{header = #diameter_header{is_request = false}}
          = Pkt,
          #state{transport = TPid}) ->
-    send(TPid, Pkt);
+    send(TPid, Pkt),
+    ok;
 
 %% Outgoing request: discard.
-outgoing(Msg, #state{dpr = {_,_}}) ->
+outgoing(Msg, #state{dpr = {_,_,_}}) ->
     invalid(false, send_after_dpr, header(Msg)).
 
 header(#diameter_packet{header = H}) ->
@@ -782,8 +837,11 @@ cea(CEA, RC, Dict0) ->
 
 post('CER' = T, RC, Pkt, S) ->
     {T, caps(S), {RC, Pkt}};
-post('DPR' = T, _, _, #state{parent = Pid}) ->
-    [fun(S) -> Pid ! {T, self()}, S end].
+post('DPR', _, _, #state{parent = Pid}) ->
+    [fun(S) -> inform_dpr(Pid), S end].
+
+inform_dpr(Pid) ->
+    Pid ! {'DPR', self()}.  %% tell watchdog to die with us
 
 rejected({capabilities_cb, _F, Reason}, T, S) ->
     rejected(Reason, T, S);
@@ -1107,10 +1165,9 @@ dpr([CB|Rest], [Reason | _] = Args, S) ->
 dpr([], [Reason | _], S) ->
     send_dpr(Reason, [], S).
 
--record(opts, {cause, timeout = ?DPA_TIMEOUT}).
+-record(opts, {cause, timeout}).
 
-send_dpr(Reason, Opts, #state{transport = TPid,
-                              dictionary = Dict,
+send_dpr(Reason, Opts, #state{dictionary = Dict,
                               service = #diameter_service{capabilities = Caps}}
                        = S) ->
     #opts{cause = Cause, timeout = Tmo}
@@ -1119,24 +1176,37 @@ send_dpr(Reason, Opts, #state{transport = TPid,
                                         transport -> ?GOAWAY;
                                         _         -> ?REBOOT
                                     end,
-                            timeout = ?DPA_TIMEOUT},
+                            timeout = dpa_timeout()},
                       Opts),
     #diameter_caps{origin_host = {OH, _},
                    origin_realm = {OR, _}}
         = Caps,
 
-    #diameter_packet{header = #diameter_header{end_to_end_id = Eid,
-                                               hop_by_hop_id = Hid}}
-        = Pkt
-        = encode(['DPR', {'Origin-Host', OH},
+    Pkt = encode(['DPR', {'Origin-Host', OH},
                          {'Origin-Realm', OR},
                          {'Disconnect-Cause', Cause}],
                  Dict),
-    incr(send, Pkt, Dict),
+    send_dpr(false, Pkt, Tmo, S).
+
+%% send_dpr/4
+
+send_dpr(X,
+         #diameter_packet{header = #diameter_header{end_to_end_id = Eid,
+                                                    hop_by_hop_id = Hid}}
+         = Pkt,
+         Tmo,
+         #state{transport = TPid,
+                dictionary = Dict}
+         = S) ->
+    %% Only count DPR sent by the service: explicit DPR is counted in
+    %% the same way as other explicitly sent requests.
+    X orelse incr(send, Pkt, Dict),
     send(TPid, Pkt),
     dpa_timer(Tmo),
     ?LOG(send, 'DPR'),
-    S#state{dpr = {Hid, Eid}}.
+    S#state{dpr = {X, Hid, Eid}}.
+
+%% opt/2
 
 opt({timeout, Tmo}, Rec)
   when ?IS_TIMEOUT(Tmo) ->
@@ -1159,6 +1229,14 @@ cause(N) ->
 dpa_timer(Tmo) ->
     erlang:send_after(Tmo, self(), dpa_timeout).
 
+dpa_timeout() ->
+    dpa_timeout(getr(?DPA_KEY)).
+
+dpa_timeout(undefined) ->
+    ?DPA_TIMEOUT;
+dpa_timeout(Tmo) ->
+    Tmo.
+
 %% register_everywhere/1
 %%
 %% Register a term and ensure it's not registered elsewhere. Note that
diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl
index 67715906e8..ff51c6dcf7 100644
--- a/lib/diameter/src/base/diameter_watchdog.erl
+++ b/lib/diameter/src/base/diameter_watchdog.erl
@@ -333,8 +333,9 @@ transition({shutdown = T, Pid, Reason}, #watchdog{parent = Pid,
     send(TPid, {T, self(), Reason}),
     S#watchdog{shutdown = true};
 
-%% Transport is telling us that DPA has been sent in response to DPR:
-%% its death should lead to ours.
+%% Transport is telling us that DPA has been sent in response to DPR,
+%% or that DPR has been explicitly sent: transport death should lead
+%% to ours.
 transition({'DPR', TPid}, #watchdog{transport = TPid} = S) ->
     S#watchdog{shutdown = true};
 
@@ -550,7 +551,7 @@ send_watchdog(#watchdog{pending = false,
     ?LOG(send, 'DWR'),
     S#watchdog{pending = true}.
 
-%% Dont' count encode errors since we don't expect any on DWR/DWA.
+%% Don't count encode errors since we don't expect any on DWR/DWA.
 
 %% recv/3
 
@@ -591,9 +592,10 @@ rcv('DWA', Pkt, #watchdog{transport = TPid,
 rcv(N, _, _)
   when N == 'CER';
        N == 'CEA';
-       N == 'DPR';
-       N == 'DPA' ->
+       N == 'DPR' ->
     false;
+%% DPR can be sent explicitly with diameter:call/4. Only the
+%% corresponding DPAs arrive here.
 
 rcv(_, Pkt, #watchdog{transport = TPid,
                       dictionary = Dict0,
diff --git a/lib/diameter/test/diameter_config_SUITE.erl b/lib/diameter/test/diameter_config_SUITE.erl
index a4bba42b2b..e61d3aea0e 100644
--- a/lib/diameter/test/diameter_config_SUITE.erl
+++ b/lib/diameter/test/diameter_config_SUITE.erl
@@ -157,6 +157,9 @@
          {length_errors,
           [[exit], [handle], [discard]],
           [[x]]},
+         {dpa_timeout,
+          [[0], [3000], [16#FFFFFFFF]],
+          [[infinity], [-1], [1 bsl 32], [x]]},
          {connect_timer,
           [[3000]],
           [[infinity]]},
diff --git a/lib/diameter/test/diameter_dpr_SUITE.erl b/lib/diameter/test/diameter_dpr_SUITE.erl
index f3f16b06e0..81178e2bda 100644
--- a/lib/diameter/test/diameter_dpr_SUITE.erl
+++ b/lib/diameter/test/diameter_dpr_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2012-2014. All Rights Reserved.
+%% Copyright Ericsson AB 2012-2015. All Rights Reserved.
 %%
 %% The contents of this file are subject to the Erlang Public License,
 %% Version 1.1, (the "License"); you may not use this file except in
@@ -32,6 +32,7 @@
 %% testcases
 -export([start/1,
          connect/1,
+         send_dpr/1,
          remove_transport/1,
          stop_service/1,
          check/1,
@@ -41,6 +42,7 @@
 -export([disconnect/5]).
 
 -include("diameter.hrl").
+-include("diameter_gen_base_rfc6733.hrl").
 
 %% ===========================================================================
 
@@ -51,9 +53,6 @@
 -define(CLIENT, "CLIENT").
 -define(SERVER, "SERVER").
 
--define(DICT_COMMON, ?DIAMETER_DICT_COMMON).
--define(APP_ID, ?DICT_COMMON:id()).
-
 %% Config for diameter:start_service/2.
 -define(SERVICE(Host),
         [{'Origin-Host', Host},
@@ -61,9 +60,10 @@
          {'Host-IP-Address', [?ADDR]},
          {'Vendor-Id', hd(Host)},  %% match this in disconnect/5
          {'Product-Name', "OTP/diameter"},
-         {'Acct-Application-Id', [?APP_ID]},
+         {'Acct-Application-Id', [0]},
          {restrict_connections, false},
-         {application, [{dictionary, ?DICT_COMMON},
+         {application, [{dictionary, diameter_gen_base_rfc6733},
+                        {alias, common},
                         {module, #diameter_callback{_ = false}}]}]).
 
 %% Disconnect reasons that diameter passes as the first argument of a
@@ -74,10 +74,12 @@
 -define(CAUSES, [0, rebooting, 1, busy, 2, goaway]).
 
 %% Establish one client connection for each element of this list,
-%% configured with disconnect/5 as disconnect_cb and returning the
-%% specified value.
+%% configured with disconnect/5, disconnect_cb returning the specified
+%% value.
 -define(RETURNS,
-        [[close, {dpr, [{cause, invalid}]}], [ignore, close], []]
+        [[close, {dpr, [{cause, invalid}]}],
+         [ignore, close],
+         []]
         ++ [[{dpr, [{timeout, 5000}, {cause, T}]}] || T <- ?CAUSES]).
 
 %% ===========================================================================
@@ -86,7 +88,7 @@ suite() ->
     [{timetrap, {seconds, 60}}].
 
 all() ->
-    [{group, R} || R <- ?REASONS].
+    [start, send_dpr, stop | [{group, R} || R <- ?REASONS]].
 
 %% The group determines how transports are terminated: by remove_transport,
 %% stop_service or application stop.
@@ -111,6 +113,22 @@ start(_Config) ->
     ok = diameter:start_service(?SERVER, ?SERVICE(?SERVER)),
     ok = diameter:start_service(?CLIENT, ?SERVICE(?CLIENT)).
 
+send_dpr(_Config) ->
+    LRef = ?util:listen(?SERVER, tcp),
+    Ref = ?util:connect(?CLIENT, tcp, LRef, [{dpa_timeout, 10000}]),
+    #diameter_base_DPA{'Result-Code' = 2001}
+        = diameter:call(?CLIENT,
+                        common,
+                        ['DPR', {'Origin-Host', "CLIENT.erlang.org"},
+                                {'Origin-Realm', "erlang.org"},
+                                {'Disconnect-Cause', 0}]),
+    ok =  receive  %% endure the transport dies on DPA
+              #diameter_event{service = ?CLIENT, info = {down, Ref, _, _}} ->
+                  ok
+          after 5000 ->
+                  erlang:process_info(self(), messages)
+          end.
+
 connect(Config) ->
     Pid = spawn(fun init/0),  %% process for disconnect_cb to bang
     Grp = group(Config),