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.

1 files changed, 81 insertions, 23 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) ->