Tweak/document request handler callback

The possibility of configuring an MFA as spawn_opt was added in commit
fd285079, the callback being passed an arity-0 fun to be applied in an
appropriate handler process. Replace the fun by a tuple to be passed to
diameter_traffic:request/1, to avoid passing funs between nodes when
handler processes are remote.

A list-valued spawn_opt is now equivalent to the following configured as
{spawn_opt, {Mod, spawn_local, [Opts]}}.

  spawn_local(ReqT, Opts) ->
      spawn_opt(diameter_traffic, request, [ReqT], Opts).

ReqT is passed by diameter and contains information that the callback
may want to decide where to handle the request in question (which wasn't
accessible with a fun), but this information isn't exposed in a
documented way. The intention is instead to add an own callback
implementation to make use of the information.

Note that application lookup now takes place in the watchdog process in
both the list-valued (or no configuration) and mfa-valued cases. Whether
this is good, bad, or (probably) inconsequential remains to be seen.

4 files changed, 74 insertions, 45 deletions

diff --git a/lib/diameter/doc/src/diameter.xml b/lib/diameter/doc/src/diameter.xml
index dfa4c803ed..b98e55d2bf 100644
--- a/lib/diameter/doc/src/diameter.xml
+++ b/lib/diameter/doc/src/diameter.xml
@@ -1,7 +1,9 @@
 <?xml version="1.0" encoding="utf-8" ?>
 <!DOCTYPE erlref SYSTEM "erlref.dtd" [
-  <!ENTITY spawn_opt
+  <!ENTITY spawn_opt2
   '<seealso marker="erts:erlang#spawn_opt-2">erlang:spawn_opt/2</seealso>'>
+  <!ENTITY spawn_opt5
+  '<seealso marker="erts:erlang#spawn_opt-5">erlang:spawn_opt/5</seealso>'>
   <!ENTITY nodes
   '<seealso marker="erts:erlang#nodes-0">erlang:nodes/0</seealso>'>
   <!ENTITY make_ref
@@ -21,7 +23,7 @@
 
 <copyright>
 <year>2011</year>
-<year>2017</year>
+<year>2019</year>
 <holder>Ericsson AB. All Rights Reserved.</holder>
 </copyright>
 <legalnotice>
@@ -1384,12 +1386,22 @@ the same peer.</p>
 </item>
 
 <tag>
-<marker id="spawn_opt"/><c>{spawn_opt, [term()]}</c></tag>
+<marker id="spawn_opt"/><c>{spawn_opt, [term()] | {M,F,A}}</c></tag>
 <item>
 <p>
-Options passed to &spawn_opt; when spawning a process for an
-incoming Diameter request.
-Options <c>monitor</c> and <c>link</c> are ignored.</p>
+An options list passed to &spawn_opt2; to spawn a handler process for an
+incoming Diameter request on the local node, or an MFA that returns
+the pid of a handler process.</p>
+
+<p>
+Options <c>monitor</c> and <c>link</c> are ignored in the list-valued
+case.
+An MFA is applied with an additional term prepended to its argument
+list, and should return either the pid of the handler process that
+invokes <c>diameter_traffic:request/1</c> on the term in order to
+process the request, or the atom <c>discard</c>.
+The handler process need not be local, but diameter must be started on
+the remote node.</p>
 
 <p>
 Defaults to the empty list.</p>
diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl
index b90b794611..7f172e1fa1 100644
--- a/lib/diameter/src/base/diameter.erl
+++ b/lib/diameter/src/base/diameter.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2017. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2019. 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.
@@ -365,7 +365,7 @@ call(SvcName, App, Message) ->
     | {connect_timer, 'Unsigned32'()}
     | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}}
     | {watchdog_config, [{okay|suspect, non_neg_integer()}]}
-    | {spawn_opt, list()}.
+    | {spawn_opt, list() | mfa()}.
 
 %% Options passed to start_service/2
 
diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl
index d2856ae530..e9acb5c0e8 100644
--- a/lib/diameter/src/base/diameter_traffic.erl
+++ b/lib/diameter/src/base/diameter_traffic.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2013-2017. All Rights Reserved.
+%% Copyright Ericsson AB 2013-2019. 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.
@@ -44,6 +44,7 @@
 
 %% internal
 -export([send/1,    %% send from remote node
+         request/1, %% process request in handler process
          init/1]).  %% monitor process start
 
 -include_lib("diameter/include/diameter.hrl").
@@ -232,7 +233,7 @@ incr_rc(Dir, Pkt, TPid, MsgDict, AppDict, Dict0) ->
 -spec receive_message(pid(), Route, #diameter_packet{}, module(), RecvData)
    -> pid()     %% request handler
     | boolean() %% answer, known request or not
-    | discard   %% request discarded by MFA
+    | discard   %% request discarded
  when Route :: {Handler, RequestRef, TPid}
              | Ack,
       RecvData :: {[SpawnOpt], #recvdata{}},
@@ -252,7 +253,8 @@ receive_message(TPid, Route, Pkt, Dict0, RecvData) ->
 recv(true, Ack, TPid, Pkt, Dict0, T)
   when is_boolean(Ack) ->
     {Opts, RecvData} = T,
-    spawn_request(Ack, TPid, Pkt, Dict0, RecvData, Opts);
+    AppT = find_app(TPid, Pkt, RecvData),
+    ack(Ack, TPid, spawn_request(AppT, Opts, Ack, TPid, Pkt, Dict0, RecvData));
 
 %% ... answer to known request ...
 recv(false, {Pid, Ref, TPid}, _, Pkt, Dict0, _) ->
@@ -274,58 +276,65 @@ recv(false, false, TPid, Pkt, _, _) ->
     incr(TPid, {{unknown, 0}, recv, discarded}),
     false.
 
-%% spawn_request/6
+%% spawn_request/7
 
-%% An MFA should return a pid() or the atom 'discard'. The latter
-%% results in an acknowledgment back to the transport process when
-%% appropriate, to ensure that send/recv callbacks can count
-%% outstanding requests. Acknowledgement is implicit if the
+spawn_request(false, _, _, _, _, _, _) ->  %% no transport
+    discard;
+
+%% An MFA should return the pid() of a process in which the argument
+%% fun in applied, or the atom 'discard' if the fun is not applied.
+%% The latter results in an acknowledgment back to the transport
+%% process when appropriate, to ensure that send/recv callbacks can
+%% count outstanding requests. Acknowledgement is implicit if the
 %% handler process dies (in a handle_request callback for example).
-spawn_request(Ack, TPid, Pkt, Dict0, RecvData, {M,F,A}) ->
-    ReqF = fun() ->
-                   ack(Ack, TPid, recv_request(Ack, TPid, Pkt, Dict0, RecvData))
-           end,
-    ack(Ack, TPid, apply(M, F, [ReqF | A]));
+spawn_request(AppT, {M,F,A}, Ack, TPid, Pkt, Dict0, RecvData) ->
+    %% Term to pass to request/1 in an appropriate process.
+    ReqT = {Pkt, AppT, Ack, TPid, Dict0, RecvData},
+    apply(M, F, [ReqT | A]);
 
 %% A spawned process acks implicitly when it dies, so there's no need
 %% to handle 'discard'.
-spawn_request(Ack, TPid, Pkt, Dict0, RecvData, Opts) ->
+spawn_request(AppT, Opts, Ack, TPid, Pkt, Dict0, RecvData) ->
     spawn_opt(fun() ->
-                      recv_request(Ack, TPid, Pkt, Dict0, RecvData)
+                      recv_request(Ack, TPid, Pkt, Dict0, RecvData, AppT)
               end,
               Opts).
 
+%% request/1
+%%
+%% Called from a handler process chosen by a transport spawn_opt MFA
+%% to process an incoming request.
+
+request({Pkt, AppT, Ack, TPid, Dict0, RecvData} = _ReqT) ->
+    ack(Ack, TPid, recv_request(Ack, TPid, Pkt, Dict0, RecvData, AppT)).
+
 %% ack/3
 
 ack(Ack, TPid, RC) ->
-    RC == discard andalso Ack andalso (TPid ! {send, false}),
+    RC == discard
+        andalso Ack
+        andalso (TPid ! {send, false}),
     RC.
 
 %% ---------------------------------------------------------------------------
-%% recv_request/5
+%% recv_request/6
 %% ---------------------------------------------------------------------------
 
 -spec recv_request(Ack :: boolean(),
                    TPid :: pid(),
                    #diameter_packet{},
                    Dict0 :: module(),
-                   #recvdata{})
+                   #recvdata{},
+                   AppT :: {#diameter_app{}, #diameter_caps{}}
+                         | #diameter_caps{}) %% no suitable app
    -> ok        %% answer was sent
-    | discard   %% or not
-    | false.    %% no transport
-
-recv_request(Ack,
-             TPid,
-             #diameter_packet{header = #diameter_header{application_id = Id}}
-             = Pkt,
-             Dict0,
-             #recvdata{peerT = PeerT,
-                       apps = Apps,
-                       counters = Count}
-             = RecvData) ->
+    | discard.  %% or not
+
+recv_request(Ack, TPid, Pkt, Dict0, RecvData, AppT) ->
     Ack andalso (TPid ! {handler, self()}),
-    case diameter_service:find_incoming_app(PeerT, TPid, Id, Apps) of
+    case AppT of
         {#diameter_app{id = Aid, dictionary = AppDict} = App, Caps} ->
+            Count = RecvData#recvdata.counters,
             Count andalso incr(recv, Pkt, TPid, AppDict),
             DecPkt = decode(Aid, AppDict, RecvData, Pkt),
             Count andalso incr_error(recv, DecPkt, TPid, AppDict),
@@ -349,11 +358,20 @@ recv_request(Ack,
                         Dict0,
                         RecvData,
                         DecPkt,
-                        [[]]);
-        false = No ->  %% transport has gone down
-            No
+                        [[]])
     end.
 
+%% find_app/3
+%%
+%% Lookup the application of a received Diameter request on the node
+%% on which it's received.
+
+find_app(TPid,
+         #diameter_packet{header = #diameter_header{application_id = Id}},
+         #recvdata{peerT = PeerT,
+                   apps = Apps}) ->
+    diameter_service:find_incoming_app(PeerT, TPid, Id, Apps).
+
 %% decode/4
 
 decode(Id, Dict, #recvdata{codec = Opts}, Pkt) ->
diff --git a/lib/diameter/test/diameter_traffic_SUITE.erl b/lib/diameter/test/diameter_traffic_SUITE.erl
index 434aef01dd..47b00c25a2 100644
--- a/lib/diameter/test/diameter_traffic_SUITE.erl
+++ b/lib/diameter/test/diameter_traffic_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2018. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2019. 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.
@@ -539,8 +539,7 @@ add_transports(Config) ->
                         ++ [{unordered, unordered()} || T == sctp],
                         [{capabilities_cb, fun capx/2},
                          {pool_size, 8}
-                         | server_apps()]
-                        ++ [{spawn_opt, {erlang, spawn, []}} || CS]),
+                         | server_apps()]),
     Cs = [?util:connect(CN,
                         [T, {sender, CS} | client_opts(T)],
                         LRef,