Allow multiple transport_module and transport_config

Transports are started one after the other if a connection is
not established with the timeout that can now be specified
with transport_config. For example, try an SCTP connect first,
a TCP connect if it doesn't succeed.

3 files changed, 210 insertions, 53 deletions

diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl
index 336f0c1f2d..6703841f80 100644
--- a/lib/diameter/src/base/diameter.erl
+++ b/lib/diameter/src/base/diameter.erl
@@ -312,6 +312,7 @@ call(SvcName, App, Message) ->
 -type transport_opt()
    :: {transport_module, atom()}
     | {transport_config, any()}
+    | {transport_timeout, non_neg_integer() | infinity}
     | {applications, [app_alias()]}
     | {capabilities, [capability()]}
     | {capabilities_cb, evaluable()}
diff --git a/lib/diameter/src/base/diameter_peer.erl b/lib/diameter/src/base/diameter_peer.erl
index 3e78c4caef..bfd59bee8e 100644
--- a/lib/diameter/src/base/diameter_peer.erl
+++ b/lib/diameter/src/base/diameter_peer.erl
@@ -27,7 +27,7 @@
          up/2]).
 
 %% ... and the stack.
--export([start/3,
+-export([start/1,
          send/2,
          close/1,
          abort/1,
@@ -57,6 +57,11 @@
 %% Server state.
 -record(state, {id = now()}).
 
+%% Default transport_module/config.
+-define(DEFAULT_TMOD, diameter_tcp).
+-define(DEFAULT_TCFG, []).
+-define(DEFAULT_TTMO, infinity).
+
 %%% ---------------------------------------------------------------------------
 %%% # notify/2
 %%% ---------------------------------------------------------------------------
@@ -65,12 +70,95 @@ notify(SvcName, T) ->
     rpc:abcast(nodes(), ?SERVER, {notify, SvcName, T}).
 
 %%% ---------------------------------------------------------------------------
-%%% # start/3
+%%% # start/1
 %%% ---------------------------------------------------------------------------
 
-start(T, Opts, #diameter_service{} = Svc) ->
-    {Mod, Cfg} = split_transport(Opts),
-    apply(Mod, start, [T, Svc, Cfg]).
+%% Initial start.
+start({T, Opts, #diameter_service{} = Svc}) ->
+    start(T, Svc, pair(Opts, [], []), []);
+
+%% Subsequent start.
+start({#diameter_service{} = Svc, Tmo, {{T, _, Cfg}, Ms, Rest, Errs}}) ->
+    start(T, Ms, Cfg, Svc, Tmo, Rest, Errs).
+
+%% pair/3
+%%
+%% Pair transport modules with config.
+
+%% Another transport_module: accumulate it.
+pair([{transport_module, M} | Rest], Mods, Acc) ->
+    pair(Rest, [M|Mods], Acc);
+
+%% Another transport_config: accumulate another tuple.
+pair([{transport_config = T, C} | Rest], Mods, Acc) ->
+    pair([{T, C, ?DEFAULT_TTMO} | Rest], Mods, Acc);
+pair([{transport_config, C, Tmo} | Rest], Mods, Acc) ->
+    pair(Rest, [], acc({Mods, C, Tmo}, Acc));
+
+pair([_ | Rest], Mods, Acc) ->
+    pair(Rest, Mods, Acc);
+
+%% No transport_module or transport_config: defaults.
+pair([], [], []) ->  
+    [{[?DEFAULT_TMOD], ?DEFAULT_TCFG, ?DEFAULT_TTMO}];
+
+%% One transport_module, one transport_config.
+pair([], [M], [{[], Cfg, Tmo}]) ->
+    [{[M], Cfg, Tmo}];
+
+%% Trailing transport_module: default transport_config.
+pair([], [_|_] = Mods, Acc) ->
+    lists:reverse(acc({Mods, ?DEFAULT_TCFG, ?DEFAULT_TTMO}, Acc));
+
+pair([], [], Acc) ->
+    lists:reverse(def(Acc)).
+
+%% acc/2
+
+acc(T, Acc) ->
+    [T | def(Acc)].
+
+%% def/1
+%%
+%% Default module of previous pair if none were specified.
+
+def([{[], Cfg, Tmo} | Acc]) ->
+    [{[?DEFAULT_TMOD], Cfg, Tmo} | Acc];
+def(Acc) ->
+    Acc.
+
+%% start/4
+
+start(T, Svc, [{Ms, Cfg, Tmo} | Rest], Errs) ->
+    start(T, Ms, Cfg, Svc, Tmo, Rest, Errs);
+
+%% One transport: return the bare error for backwards compatibility.
+start(_, _, [], [Err]) ->
+    {Err};
+
+%% Or not: return list of errors.
+start(_, _, [], Errs) ->
+    {{error, Errs}}.
+
+%% start/7
+
+start(T, [], _, Svc, _, Rest, Errs) ->
+    start(T, Svc, Rest, Errs);
+
+start(T, [M|Ms], Cfg, Svc, Tmo, Rest, Errs) ->
+    case start(M, [T, Svc, Cfg]) of
+        {ok, TPid} ->
+            {TPid, [], Tmo, {{T, M, Cfg}, Ms, Rest, Errs}};
+        {ok, TPid, [_|_] = Addrs} ->
+            {TPid, Addrs, Tmo, {{T, M, Cfg}, Ms, Rest, Errs}};
+        E ->
+            start(T, Ms, Cfg, Svc, Tmo, Rest, [E|Errs])
+    end.
+
+%% start/2
+
+start(Mod, Args) ->
+    apply(Mod, start, Args).
 
 %%% ---------------------------------------------------------------------------
 %%% # up/[12]
@@ -204,21 +292,6 @@ bang(undefined = No, _) ->
 bang(Pid, T) ->
     Pid ! T.
 
-%% split_transport/1
-%%
-%% Split options into transport module, transport config and
-%% remaining options.
-
-split_transport(Opts) ->
-    {[M,C], _} = proplists:split(Opts, [transport_module,
-                                        transport_config]),
-    {value(M, diameter_tcp), value(C, [])}.
-
-value([{_,V}], _) ->
-    V;
-value([], V) ->
-    V.
-
 %% call/1
 
 call(Request) ->
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index 99644814d2..d5e2d690cf 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -54,6 +54,12 @@
 -define(NO_INBAND_SECURITY, 0).
 -define(TLS, 1).
 
+%% Keys in process dictionary.
+-define(CB_KEY, cb).       %% capabilities callback
+-define(DWA_KEY, dwa).     %% outgoing DWA
+-define(Q_KEY, q).         %% transport start queue
+-define(START_KEY, start). %% start of connected transport
+
 %% A 2xxx series Result-Code. Not necessarily 2001.
 -define(IS_SUCCESS(N), 2 == (N) div 1000).
 
@@ -143,18 +149,17 @@ init(T) ->
     proc_lib:init_ack({ok, self()}),
     gen_server:enter_loop(?MODULE, [], i(T)).
 
-i({WPid, T, Opts, #diameter_service{capabilities = Caps} = Svc0}) ->
-    putr(dwa, dwa(Caps)),
+i({WPid, T, Opts, #diameter_service{capabilities = Caps} = Svc}) ->
+    putr(?DWA_KEY, dwa(Caps)),
     {M, Ref} = T,
     {[Ts], Rest} = proplists:split(Opts, [capabilities_cb]),
-    putr(capabilities_cb, {Ref, [F || {_,F} <- Ts]}),
-    {ok, TPid, Svc} = start_transport(T, Rest, Svc0),
-    erlang:monitor(process, TPid),
+    putr(?CB_KEY, {Ref, [F || {_,F} <- Ts]}),
     erlang:monitor(process, WPid),
+    {TPid, Addrs} = start_transport(T, Rest, Svc),
     #state{parent = WPid,
            transport = TPid,
            mode = M,
-           service = Svc}.
+           service = svc(Svc, Addrs)}.
 %% The transport returns its local ip addresses so that different
 %% transports on the same service can use different local addresses.
 %% The local addresses are put into Host-IP-Address avps here when
@@ -164,18 +169,53 @@ i({WPid, T, Opts, #diameter_service{capabilities = Caps} = Svc0}) ->
 %% watchdog start (start/2) succeeds regardless so as not to crash the
 %% service.
 
-start_transport(T, Opts, Svc) ->
-    case diameter_peer:start(T, Opts, Svc) of
-        {ok, TPid} ->
-            {ok, TPid, Svc};
-        {ok, TPid, [_|_] = Addrs} ->
-            #diameter_service{capabilities = Caps0} = Svc,
-            Caps = Caps0#diameter_caps{host_ip_address = Addrs},
-            {ok, TPid, Svc#diameter_service{capabilities = Caps}};
-        No ->
+start_transport(T, Opts, #diameter_service{capabilities = Caps} = Svc) ->
+    Addrs0 = Caps#diameter_caps.host_ip_address,
+    start_transport(Addrs0, {T, Opts, Svc}).
+
+start_transport(Addrs0, T) ->
+    case diameter_peer:start(T) of
+        {TPid, Addrs, Tmo, Data} ->
+            erlang:monitor(process, TPid),
+            q_next(TPid, Addrs0, Tmo, Data),
+            {TPid, addrs(Addrs, Addrs0)};
+        {No} ->
             exit({shutdown, No})
     end.
 
+addrs([], Addrs0) ->
+    Addrs0;
+addrs(Addrs, _) ->
+    Addrs.
+
+svc(Svc, []) ->
+    Svc;
+svc(Svc, Addrs) ->
+    readdr(Svc, Addrs).
+
+readdr(#diameter_service{capabilities = Caps0} = Svc, Addrs) ->
+    Caps = Caps0#diameter_caps{host_ip_address = Addrs},
+    Svc#diameter_service{capabilities = Caps}.
+
+%% The 4-tuple Data returned from diameter_peer:start/1 identifies the
+%% transport module/config use to start the transport process in
+%% question as well as any alternates to try if a connection isn't
+%% established within Tmo.
+q_next(TPid, Addrs0, Tmo, {_,_,_,_} = Data) ->
+    send_after(Tmo, {connection_timeout, TPid}),
+    putr(?Q_KEY, {Addrs0, Tmo, Data}).
+
+%% Connection has been established: retain the started module/config
+%% in the process dictionary.
+keep_transport() ->
+    {_, _, {{_,_,_} = T, _, _, _}} = eraser(?Q_KEY),
+    putr(?START_KEY, T).
+
+send_after(infinity, _) ->
+    ok;
+send_after(Tmo, T) ->
+    erlang:send_after(Tmo, self(), T).
+
 %% handle_call/3
 
 handle_call(_, _, State) ->
@@ -240,25 +280,48 @@ eraser(Key) ->
 
 %% Connection to peer.
 transition({diameter, {TPid, connected, Remote}},
-           #state{state = PS,
+           #state{transport = TPid,
+                  state = PS,
                   mode = M}
            = S) ->
     'Wait-Conn-Ack' = PS,  %% assert
     connect = M,           %%
-    send_CER(S#state{mode = {M, Remote},
-                     transport = TPid});
+    keep_transport(),
+    send_CER(S#state{mode = {M, Remote}});
 
 %% Connection from peer.
 transition({diameter, {TPid, connected}},
-           #state{state = PS,
+           #state{transport = TPid,
+                  state = PS,
                   mode = M,
                   parent = Pid}
            = S) ->
     'Wait-Conn-Ack' = PS,  %% assert
     accept = M,            %%
+    keep_transport(),
     Pid ! {accepted, self()},
-    start_timer(S#state{state = recv_CER,
-                        transport = TPid});
+    start_timer(S#state{state = recv_CER});
+
+%% Connection established after receiving a connection_timeout
+%% message. This may be followed by an incoming message which arrived
+%% before the transport was killed and this can't be distinguished
+%% from one from the transport that's been started to replace it.
+transition({diameter, {_, connected}}, _) ->
+    {stop, connection_timeout};
+transition({diameter, {_, connected, _}}, _) ->
+    {stop, connection_timeout};
+
+%% Connection has timed out: start an alternate.
+transition({connection_timeout = T, TPid},
+           #state{transport = TPid,
+                  state = 'Wait-Conn-Ack'}
+           = S) ->
+    exit(TPid, {shutdown, T}),
+    start_next(S);
+
+%% Connect timeout after connection or alternate start: ignore.
+transition({connection_timeout, _}, _) ->
+    ok;
 
 %% Incoming message from the transport.
 transition({diameter, {recv, Pkt}}, S) ->
@@ -305,14 +368,21 @@ transition({resolve_port, _Pid} = T, #state{transport = TPid}) ->
     TPid ! T,
     ok;
 
-%% Parent or transport has died.
-transition({'DOWN', _, process, P, _},
-           #state{parent = Pid,
-                  transport = TPid})
-  when P == Pid;
-       P == TPid ->
+%% Parent has died.
+transition({'DOWN', _, process, WPid, _},
+           #state{parent = WPid}) ->
     stop;
 
+%% Transport has died before connection timeout.
+transition({'DOWN', _, process, TPid, _},
+           #state{transport = TPid}
+           = S) ->
+    start_next(S);
+
+%% Transport has died after connection timeout.
+transition({'DOWN', _, process, _, _}, _) ->
+    ok;
+
 %% State query.
 transition({state, Pid}, #state{state = S, transport = TPid}) ->
     Pid ! {self(), [S, TPid]},
@@ -320,6 +390,19 @@ transition({state, Pid}, #state{state = S, transport = TPid}) ->
 
 %% Crash on anything unexpected.
 
+%% start_next/1
+
+start_next(#state{service = Svc0} = S) ->
+    case getr(?Q_KEY) of
+        {Addrs0, Tmo, Data} ->
+            Svc = readdr(Svc0, Addrs0),
+            {TPid, Addrs} = start_transport(Addrs0, {Svc, Tmo, Data}),
+            S#state{transport = TPid,
+                    service = svc(Svc, Addrs)};
+        undefined ->
+            stop
+    end.
+
 %% send_CER/1
 
 send_CER(#state{mode = {connect, Remote},
@@ -649,7 +732,7 @@ rc([RC|_]) ->
 %% answer/2
 
 answer('DWR', _) ->
-    getr(dwa);
+    getr(?DWA_KEY);
 
 answer(Name, #state{service = #diameter_service{capabilities = Caps}}) ->
     a(Name, Caps).
@@ -749,15 +832,15 @@ caps(#state{service = Svc}) ->
 %% caps_cb/1
 
 caps_cb(Caps) ->
-    {Ref, Ts} = eraser(capabilities_cb),
-    ccb(Ts, [Ref, Caps]).
+    {Ref, Ts} = eraser(?CB_KEY),
+    caps_cb(Ts, [Ref, Caps]).
 
-ccb([], _) ->
+caps_cb([], _) ->
     ok;
-ccb([F | Rest], T) ->
+caps_cb([F | Rest], T) ->
     case diameter_lib:eval([F|T]) of
         ok ->
-            ccb(Rest, T);
+            caps_cb(Rest, T);
         N when ?IS_SUCCESS(N) ->  %% 2xxx result code: accept immediately
             N;
         Res ->