Merge branch 'anders/diameter/message_cb/OTP-14486' into maint

* anders/diameter/message_cb/OTP-14486:
  Add simple message_cb to example server
  Fix inappropriate message callbacks

2 files changed, 41 insertions, 8 deletions

diff --git a/lib/diameter/examples/code/node.erl b/lib/diameter/examples/code/node.erl
index 246be4194b..fc5830f8e2 100644
--- a/lib/diameter/examples/code/node.erl
+++ b/lib/diameter/examples/code/node.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2016. 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.
@@ -30,6 +30,8 @@
          connect/2,
          stop/1]).
 
+-export([message/3]).
+
 -type protocol()
    :: tcp | sctp.
 
@@ -128,6 +130,8 @@ stop(Name) ->
 server_opts({T, Addr, Port}) ->
     [{transport_module, tmod(T)},
      {transport_config, [{reuseaddr, true},
+                         {sender, true},
+                         {message_cb, [fun ?MODULE:message/3, 0]},
                          {ip, addr(Addr)},
                          {port, Port}]}];
 
@@ -173,3 +177,26 @@ addr(loopback) ->
     {127,0,0,1};
 addr(A) ->
     A.
+
+%% ---------------------------------------------------------------------------
+
+%% message/3
+%%
+%% Simple message callback that limits the number of concurrent
+%% requests on the peer connection in question.
+
+%% Incoming request.
+message(recv, <<_:32, 1:1, _/bits>> = Bin, N) ->
+    [Bin, N < 32, fun ?MODULE:message/3, N+1];
+
+%% Outgoing request.
+message(ack, <<_:32, 1:1, _/bits>>, _) ->
+    [];
+
+%% Incoming answer or request discarded.
+message(ack, _, N) ->
+    [N =< 32, fun ?MODULE:message/3, N-1];
+
+%% Outgoing message or incoming answer.
+message(_, Bin, _) ->
+    [Bin].
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index 1b0dc417e5..e43b3f54cf 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -542,11 +542,11 @@ put_route(Pid) ->
     MRef = monitor(process, Pid),
     put(Pid, MRef).
 
-%% get_route/2
+%% get_route/3
 
-%% incoming answer
-get_route(_, #diameter_packet{header = #diameter_header{is_request = false}}
-             = Pkt) ->
+%% Incoming answer.
+get_route(_, _, #diameter_packet{header = #diameter_header{is_request = false}}
+                = Pkt) ->
     Seqs = diameter_codec:sequence_numbers(Pkt),
     case erase(Seqs) of
         {Pid, Ref, MRef} ->
@@ -557,8 +557,14 @@ get_route(_, #diameter_packet{header = #diameter_header{is_request = false}}
             false
     end;
 
-%% incoming request
-get_route(Ack, _) ->
+%% Requests answered here ...
+get_route(_, N, _)
+  when N == 'CER';
+       N == 'DPR' ->
+    false;
+
+%% ... or not.
+get_route(Ack, _, _) ->
     Ack.
 
 %% erase_route/1
@@ -650,7 +656,7 @@ encode(Rec, Opts, Dict) ->
 %% incoming/2
 
 incoming({recv = T, Name, Pkt}, #state{parent = Pid, ack = Ack} = S) ->
-    Pid ! {T, self(), get_route(Ack, Pkt), Name, Pkt},
+    Pid ! {T, self(), get_route(Ack, Name, Pkt), Name, Pkt},
     rcv(Name, Pkt, S);
 
 incoming(#diameter_header{is_request = R}, #state{transport = TPid,