Allow DPR to be sent with diameter:call/4

DPR is sent by diameter at application shutdown, service stop, or
transport removal. It has been possible to send the request with
diameter:call/4, but the answer was discarded, instead of the transport
process being terminated. This commit causes DPR to be handled in the
same way regardless of whether it's sent by diameter or by
diameter:call/4.

Note that the behaviour subsequent to DPA is unchanged. In particular,
in the connecting case, the closed connection will be reestablished
after a connect_timer expiry unless the transport is removed. The more
probable use case is the listening case, to disconnect a single peer
associated with a listening transport.

2 files changed, 89 insertions, 29 deletions

diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index 434524040f..dcfcb63808 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -108,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}).
 
@@ -214,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}).
@@ -228,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) ->
@@ -335,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,
@@ -400,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.
@@ -537,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) ->
@@ -603,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}
@@ -611,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};
 
@@ -652,23 +676,42 @@ 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;
+
 %% 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}) ->
@@ -784,8 +827,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);
@@ -1111,8 +1157,7 @@ dpr([], [Reason | _], S) ->
 
 -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}
@@ -1127,18 +1172,31 @@ send_dpr(Reason, Opts, #state{transport = TPid,
                    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) ->
diff --git a/lib/diameter/src/base/diameter_watchdog.erl b/lib/diameter/src/base/diameter_watchdog.erl
index b7f2d24941..6d7f6d7675 100644
--- a/lib/diameter/src/base/diameter_watchdog.erl
+++ b/lib/diameter/src/base/diameter_watchdog.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2014. All Rights Reserved.
+%% Copyright Ericsson AB 2010-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
@@ -332,8 +332,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};
 
@@ -549,7 +550,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
 
@@ -590,9 +591,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,