Merge branch 'anders/diameter/transport/ERL-332'

* anders/diameter/transport/ERL-332: (35 commits)
  Capitulate on SCTP vs sparc-sun-solaris2.10
  Remove obsolete traffic testcase
  Fix dialyzer warnings
  Remove client/server string decode from traffic suite
  Add diameter_sctp option packet
  Add diameter_sctp send/recv callbacks
  Let diameter_tcp send/recv callbacks deal in diameter_packet
  Randomly select traffic testcases
  Exercise diameter_tcp message callbacks in traffic suite
  Exercise diameter_{tcp,sctp} sender in traffic suite
  Remove upgrade from diameter_traffic
  Add diameter_tcp send/recv callbacks
  Make diameter_{tcp,sctp} sender configurable
  Remove upgrade from diameter_sctp; tweak diameter_tcp to match
  Fix incomprehensible dialyzer warning
  Simplify acks to transport processes
  Strip throttling callbacks from diameter_tcp
  Deal with (another) SCTP association id quirk on Solaris
  Use binary:copy/2 when generating largish data in test suites
  Deal with SCTP association id quirk on Solaris
  ...

8 files changed, 510 insertions, 361 deletions

diff --git a/lib/diameter/test/diameter_capx_SUITE.erl b/lib/diameter/test/diameter_capx_SUITE.erl
index fdeff96a58..51b6c1d7f2 100644
--- a/lib/diameter/test/diameter_capx_SUITE.erl
+++ b/lib/diameter/test/diameter_capx_SUITE.erl
@@ -433,7 +433,7 @@ server_reject(Config, F, RC) ->
             ?fail({LRef, OH})
     end.
 
-%% cliient_closed/4
+%% client_closed/4
 
 client_closed(Config, Host, F, RC) ->
     true = diameter:subscribe(?CLIENT),
diff --git a/lib/diameter/test/diameter_examples_SUITE.erl b/lib/diameter/test/diameter_examples_SUITE.erl
index e4ed2b227d..fad54d62b2 100644
--- a/lib/diameter/test/diameter_examples_SUITE.erl
+++ b/lib/diameter/test/diameter_examples_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2013-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2013-2017. 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.
diff --git a/lib/diameter/test/diameter_gen_sctp_SUITE.erl b/lib/diameter/test/diameter_gen_sctp_SUITE.erl
index 79db39ca45..d76d2bdbd3 100644
--- a/lib/diameter/test/diameter_gen_sctp_SUITE.erl
+++ b/lib/diameter/test/diameter_gen_sctp_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2017. 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.
@@ -33,8 +33,8 @@
          end_per_suite/1]).
 
 %% testcases
--export([send_not_from_controlling_process/1,
-         send_from_multiple_clients/1, send_from_multiple_clients/0,
+-export([send_one_from_many/1, send_one_from_many/0,
+         send_many_from_one/1, send_many_from_one/0,
          receive_what_was_sent/1]).
 
 -include_lib("kernel/include/inet_sctp.hrl").
@@ -45,16 +45,24 @@
 %% Open sockets on the loopback address.
 -define(ADDR, {127,0,0,1}).
 
-%% Snooze, nap, siesta.
--define(SLEEP(T), receive after T -> ok end).
-
 %% An indescribably long number of milliseconds after which everthing
 %% that should have happened has.
 -define(FOREVER, 2000).
 
+%% How many milliseconds to tolerate between the fastest and slowest
+%% turnaround times.
+-define(VARIANCE, 100).
+
 %% The first byte in each message we send as a simple guard against
 %% not receiving what was sent.
--define(MAGIC, 42).
+-define(MAGIC, 0).
+
+%% Requested number of inbound/outbound streams.
+-define(STREAMS, 5).
+
+%% Success for send_multiple. Match in each testcase rather than in
+%% send_multiple itself for a better failure in common_test.
+-define(OK, {_, true, _, [true, true], [], _}).
 
 %% ===========================================================================
 
@@ -62,8 +70,8 @@ suite() ->
     [{timetrap, {seconds, 10}}].
 
 all() ->
-    [send_not_from_controlling_process,
-     send_from_multiple_clients,
+    [send_one_from_many,
+     send_many_from_one,
      receive_what_was_sent].
 
 init_per_suite(Config) ->
@@ -81,130 +89,37 @@ end_per_suite(_Config) ->
 
 %% ===========================================================================
 
-%% send_not_from_controlling_process/1
-%%
-%% This testcase failing shows gen_sctp:send/4 hanging when called
-%% outside the controlling process of the socket in question.
-
-send_not_from_controlling_process(_) ->
-    Pids = send_not_from_controlling_process(),
-    ?SLEEP(?FOREVER),
-    try
-        [] = [{P,I} || P <- Pids, I <- [process_info(P)], I /= undefined]
-    after
-        lists:foreach(fun(P) -> exit(P, kill) end, Pids)
-    end.
-
-%% send_not_from_controlling_process/0
-%%
-%% Returns the pids of three spawned processes: a listening process, a
-%% connecting process and a sending process.
-%%
-%% The expected behaviour is that all three processes exit:
-%%
-%% - The listening process exits upon receiving an SCTP message
-%%   sent by the sending process.
-%% - The connecting process exits upon listening process exit.
-%% - The sending process exits upon gen_sctp:send/4 return.
-%%
-%% The observed behaviour is that all three processes remain alive
-%% indefinitely:
-%%
-%% - The listening process never receives the SCTP message sent
-%%   by the sending process.
-%% - The connecting process has an inet_reply message in its mailbox
-%%   as a consequence of the call to gen_sctp:send/4 call from the
-%%   sending process.
-%% - The call to gen_sctp:send/4 in the sending process doesn't return,
-%%   hanging in prim_inet:getopts/2.
-
-send_not_from_controlling_process() ->
-    FPid = self(),
-    {L, MRef} = spawn_monitor(fun() -> listen(FPid) end),
-    receive
-        {?MODULE, C, S} ->
-            demonitor(MRef, [flush]),
-            [L,C,S];
-        {'DOWN', MRef, process, _, _} = T ->
-            error(T)
-    end.
-
-%% listen/1
-
-listen(FPid) ->
-    {ok, Sock} = open(),
-    ok = gen_sctp:listen(Sock, true),
-    {ok, PortNr} = inet:port(Sock),
-    LPid = self(),
-    spawn(fun() -> connect1(PortNr, FPid, LPid) end), %% connecting process
-    Id = assoc(Sock),
-    recv(Sock, Id).
-
-%% connect1/3
-
-connect1(PortNr, FPid, LPid) ->
-    {ok, Sock} = open(),
-    ok = gen_sctp:connect_init(Sock, ?ADDR, PortNr, []),
-    Id = assoc(Sock),
-    FPid ! {?MODULE,
-            self(),
-            spawn(fun() -> send(Sock, Id) end)}, %% sending process
-    MRef = monitor(process, LPid),
-    down(MRef).  %% Waits with this as current_function.
-
-%% down/1
-
-down(MRef) ->
-    receive {'DOWN', MRef, process, _, Reason} -> Reason end.
-
-%% send/2
-
-send(Sock, Id) ->
-    ok = gen_sctp:send(Sock, Id, 0, <<0:32>>).
-
-%% ===========================================================================
-
-%% send_from_multiple_clients/0
+%% send_one_from_many/0
 %%
 %% Demonstrates sluggish delivery of messages.
 
-send_from_multiple_clients() ->
-    [{timetrap, {seconds, 60}}].
+send_one_from_many() ->
+    [{timetrap, {seconds, 30}}].
 
-send_from_multiple_clients(_) ->
-    {S, Rs} = T = send_from_multiple_clients(8, 1024),
-    Max = ?FOREVER*1000,
-    {false, [], _} = {Max < S,
-                      Rs -- [OI || {O,_} = OI <- Rs, is_integer(O)],
-                      T}.
+send_one_from_many(_) ->
+    ?OK = send_multiple(128, 1, 1024).
 
-%% send_from_multiple_clients/2
+%% send_one_from_many/2
 %%
 %% Opens a listening socket and then spawns a specified number of
-%% processes, each of which connects to the listening socket. Each
-%% connecting process then sends a message, whose size in bytes is
-%% passed as an argument, the listening process sends a reply
-%% containing the time at which the message was received, and the
-%% connecting process then exits upon reception of this reply.
+%% processes, each of which connects, sends a message, receives a
+%% reply, and exits.
 %%
 %% Returns the elapsed time for all connecting process to exit
-%% together with a list of exit reasons for the connecting processes.
-%% In the successful case a connecting process exits with the
-%% outbound/inbound transit times for the sent/received message as
-%% reason.
+%% together with a list of exit reasons. In the successful case a
+%% connecting process exits with the outbound/inbound transit times
+%% for the sent/received message as reason.
 %%
 %% The observed behaviour is that some outbound messages (that is,
 %% from a connecting process to the listening process) can take an
 %% unexpectedly long time to complete their journey. The more
-%% connecting processes, the longer the possible delay it seems.
+%% connecting processes, the longer it can take it seems.
 %%
-%% eg. (With F = fun send_from_multiple_clients/2.)
-%%
-%%     5> F(2, 1024).
+%% eg. 5> send_one_from_many(2, 1024).
 %%     {875,[{128,116},{113,139}]}
-%%     6> F(4, 1024).
+%%     6> send_one_from_many(4, 1024).
 %%     {2995290,[{2994022,250},{2994071,80},{200,130},{211,113}]}
-%%     7> F(8, 1024).
+%%     7> send_one_from_many(8, 1024).
 %%     {8997461,[{8996161,116},
 %%               {2996471,86},
 %%               {2996278,116},
@@ -213,7 +128,7 @@ send_from_multiple_clients(_) ->
 %%               {213,159},
 %%               {373,173},
 %%               {376,118}]}
-%%     8> F(8, 1024).
+%%     8> send_one_from_many(8, 1024).
 %%     {21001891,[{20999968,128},
 %%                {8997891,172},
 %%                {8997927,91},
@@ -223,120 +138,279 @@ send_from_multiple_clients(_) ->
 %%                {117,98},
 %%                {149,125}]}
 %%
-%% This turns out to have been due to SCTP resends as a consequence of
-%% the listener having an insufficient recbuf. Increasing the size
-%% solves the problem.
-%%
-
-send_from_multiple_clients(N, Sz)
-  when is_integer(N), 0 < N, is_integer(Sz), 0 < Sz ->
-    timer:tc(fun listen/2, [N, <<?MAGIC, 0:Sz/unit:8>>]).
 
-%% listen/2
-
-listen(N, Bin) ->
+send_multiple(Clients, Msgs, Sz)
+  when is_integer(Clients), 0 < Clients,
+       is_integer(Msgs), 0 < Msgs,
+       is_integer(Sz), 0 < Sz ->
+    T0 = diameter_lib:now(),
+    {S, Res} = timer:tc(fun listen/3, [Clients, Msgs, Sz]),
+    report(T0, Res),
+    Ts = lists:append(Res),
+    Outgoing = [DT || {_,{_,_,DT},{_,_,_},_} <- Ts],
+    Incoming = [DT || {_,{_,_,_},{_,_,DT},_} <- Ts],
+    Diffs = [lists:max(L) - lists:min(L) || L <- [Outgoing, Incoming]],
+    {S,
+     S < ?FOREVER*1000,
+     Diffs,
+     [D < V || V <- [?VARIANCE*1000], D <- Diffs],
+     [T || T <- Ts, [] == [T || {_,{_,_,_},{_,_,_},_} <- [T]]],
+     Res}.
+
+%% listen/3
+
+listen(Clients, Msgs, Sz) ->
     {ok, Sock} = open(),
     ok = gen_sctp:listen(Sock, true),
     {ok, PortNr} = inet:port(Sock),
 
     %% Spawn a middleman that in turn spawns N connecting processes,
     %% collects a list of exit reasons and then exits with the list as
-    %% reason. loop/3 returns when we receive this list from the
+    %% reason. accept/2 returns when we receive this list from the
     %% middleman's 'DOWN'.
 
     Self = self(),
-    Fun = fun() -> exit(connect2(Self, PortNr, Bin)) end,
-    {_, MRef} = spawn_monitor(fun() -> exit(fold(N, Fun)) end),
-    loop(Sock, MRef, Bin).
+    Fun = fun() -> exit(client(Self, PortNr, Msgs, Sz)) end,  %% start clients
+    {_, MRef} = spawn_monitor(fun() -> exit(clients(Clients, Fun)) end),
+    accept_loop(Sock, MRef).
 
-%% fold/2
+%% fclients/2
 %%
 %% Spawn N processes and collect their exit reasons in a list.
 
-fold(N, Fun) ->
+clients(N, Fun) ->
     start(N, Fun),
     acc(N, []).
 
+%% start/2
+
 start(0, _) ->
     ok;
+
 start(N, Fun) ->
     spawn_monitor(Fun),
     start(N-1, Fun).
 
+%% acc/2
+
 acc(0, Acc) ->
     Acc;
+
 acc(N, Acc) ->
     receive
         {'DOWN', _MRef, process, _, RC} ->
             acc(N-1, [RC | Acc])
     end.
 
-%% loop/3
+%% accept_loop/2
 
-loop(Sock, MRef, Bin) ->
+accept_loop(Sock, MRef) ->
+    ok = inet:setopts(Sock, [{active, once}]),
     receive
-        ?SCTP(Sock, {[#sctp_sndrcvinfo{assoc_id = Id}], B})
-          when is_binary(B) ->
-            Sz = size(Bin),
-            {Sz, Bin} = {size(B), B},  %% assert
-            ok = send(Sock, Id, mark(Bin)),
-            loop(Sock, MRef, Bin);
+        ?SCTP(Sock, {_, #sctp_assoc_change{state = comm_up,
+                                           outbound_streams = OS,
+                                           assoc_id = Id}}) ->
+            Self = self(),
+            TPid = spawn(fun() -> assoc(monitor(process, Self), Id, OS) end),
+            NewSock = peeloff(Sock, Id, TPid),
+            TPid ! {peeloff, NewSock},
+            accept_loop(Sock, MRef);
         ?SCTP(Sock, _) ->
-            loop(Sock, MRef, Bin);
+            accept_loop(Sock, MRef);
         {'DOWN', MRef, process, _, Reason} ->
-            Reason
+            Reason;
+        T ->
+            error(T)
     end.
 
-%% connect2/3
+%% assoc/3
+%%
+%% Server process that answers incoming messages as long as the parent
+%% lives.
+
+assoc(MRef, _Id, OS)
+  when is_reference(MRef) ->
+    {peeloff, Sock} = receive T -> T end,
+    recv_loop(Sock, false, sender(Sock, false, OS), MRef).
+
+%% recv_loop/4
+
+recv_loop(Sock, Id, Pid, MRef) ->
+    ok = inet:setopts(Sock, [{active, once}]),
+    recv(Sock, Id, Pid, MRef, receive T -> T end).
+
+%% recv/5
+
+%% Association id can change on a peeloff socket on some versions of
+%% Solaris.
+recv(Sock,
+     false,
+     Pid,
+     MRef,
+     ?SCTP(Sock, {[#sctp_sndrcvinfo{assoc_id = Id}], _})
+     = T) ->
+    Pid ! {assoc_id, Id},
+    recv(Sock, Id, Pid, MRef, T);
+
+recv(Sock, Id, Pid, MRef, ?SCTP(Sock, {[#sctp_sndrcvinfo{assoc_id = I}], B}))
+  when is_binary(B) ->
+    T2 = diameter_lib:now(),
+    Id = I,                      %% assert
+    <<?MAGIC, Bin/binary>> = B,  %% assert
+    {[_,_,_,Sz] = L, Bytes} = unmark(Bin),
+    Sz = size(Bin) - Bytes,      %% assert
+    <<_:Bytes/binary, Body:Sz/binary>> = Bin,
+    send(Pid, [T2|L], Body),  %% answer
+    recv_loop(Sock, Id, Pid, MRef);
+
+recv(Sock, Id, Pid, MRef, ?SCTP(Sock, _)) ->
+    recv_loop(Sock, Id, Pid, MRef);
+
+recv(_, _, _, MRef, {'DOWN', MRef, process, _, Reason}) ->
+    Reason;
+
+recv(_, _, _, _, T) ->
+    error(T).
 
-connect2(Pid, PortNr, Bin) ->
-    monitor(process, Pid),
+%% send/3
 
-    {ok, Sock} = open(),
-    ok = gen_sctp:connect_init(Sock, ?ADDR, PortNr, []),
-    Id = assoc(Sock),
+send(Pid, Header, Body) ->
+    Pid ! {send, Header, Body}.
 
-    %% T1 = time before send
-    %% T2 = time after listening process received our message
-    %% T3 = time after reply is received
+%% sender/3
+%%
+%% Start a process that sends, so as not to block the controlling process.
 
-    T1 = diameter_lib:now(),
-    ok = send(Sock, Id, Bin),
-    T2 = unmark(recv(Sock, Id)),
-    T3 = diameter_lib:now(),
-    {diameter_lib:micro_diff(T2, T1),  %% Outbound
-     diameter_lib:micro_diff(T3, T2)}. %% Inbound
+sender(Sock, Id, OS) ->
+    Pid = self(),
+    spawn(fun() -> send_loop(Sock, Id, OS, 1, monitor(process, Pid)) end).
 
-%% recv/2
+%% send_loop/5
 
-recv(Sock, Id) ->
+send_loop(Sock, Id, OS, N, MRef) ->
     receive
-        ?SCTP(Sock, {[#sctp_sndrcvinfo{assoc_id = I}], Bin})
-          when is_binary(Bin) ->
-            Id = I,   %% assert
-            Bin;
-        ?SCTP(S, _) ->
-            Sock = S, %% assert
-            recv(Sock, Id);
+        {assoc_id, I} ->
+            send_loop(Sock, I, OS, N, MRef);
+        {send, L, Body} ->
+            Stream = N rem OS,
+            ok = send(Sock, Id, Stream, mark(Body, [N, Stream | L])),
+            send_loop(Sock, Id, OS, N+1, MRef);
+        {'DOWN', MRef, process, _, _} = T ->
+            T;
         T ->
-            exit(T)
+            error(T)
     end.
 
-%% send/3
+%% peeloff/3
 
-send(Sock, Id, Bin) ->
-    gen_sctp:send(Sock, Id, 0, Bin).
+peeloff(LSock, Id, TPid) ->
+    {ok, Sock} = gen_sctp:peeloff(LSock, Id),
+    ok = gen_sctp:controlling_process(Sock, TPid),
+    Sock.
 
-%% mark/1
+%% client/4
 
-mark(Bin) ->
-    Info = term_to_binary(diameter_lib:now()),
-    <<Info/binary, Bin/binary>>.
+client(Pid, PortNr, Msgs, Sz) ->
+    monitor(process, Pid),
+    {ok, Sock} = open(),
+    ok = gen_sctp:connect_init(Sock, ?ADDR, PortNr, []),
+    recv_loop(Sock, Msgs, Sz).
+
+%% recv_loop/3
+
+recv_loop(_, 0, T) ->
+    [_,_|Acc] = T,
+    Acc;
+
+recv_loop(Sock, Msgs, T) ->
+    ok = inet:setopts(Sock, [{active, once}]),
+    {I, NewT} = recv(Sock, Msgs, T, receive X -> X end),
+    recv_loop(Sock, Msgs - I, NewT).
+
+%% recv/4
+
+recv(Sock, Msgs, Sz, ?SCTP(Sock, {_, #sctp_assoc_change{} = A})) ->
+    #sctp_assoc_change{state = comm_up,  %% assert
+                       assoc_id = Id,
+                       outbound_streams = OS}
+        = A,
+    true = is_integer(Sz),  %% assert
+    send_n(Msgs, sender(Sock, Id, OS), Sz),
+    {0, [Id, OS]};
+
+recv(Sock, _, T, ?SCTP(Sock, {[#sctp_sndrcvinfo{assoc_id = Id}], Bin})) ->
+    T4 = diameter_lib:now(),
+    [Id, OS | Acc] = T,
+    {1, [Id, OS, stat(T4, Bin) | Acc]};
+
+recv(Sock, _, T, ?SCTP(Sock, _)) ->
+    {0, [_,_|_] = T};
+
+recv(_, _, _, T) ->
+    error(T).
+
+%% send_n/3
+%%
+%% Send messages to the server from dedicated processes.
+
+send_n(0, _, _) ->
+    ok;
 
+send_n(N, Pid, Sz) ->
+    M = rand:uniform(255),
+    send(Pid, [Sz], binary:copy(<<M>>, Sz)),
+    send_n(N-1, Pid, Sz).
+
+%% send/4
+
+send(Sock, Id, Stream, Bin) ->
+    case gen_sctp:send(Sock, Id, Stream, <<?MAGIC, Bin/binary>>) of
+        {error, eagain} ->
+            send(Sock, Id, Stream, Bin);
+        RC ->
+            RC
+    end.
+
+%% stat/2
+
+stat(T4, <<?MAGIC, Bin/binary>>) ->
+    %% T1 = time at send
+    %% T2 = time at reception by server
+    %% T3 = time at reception by server's sender
+    %% T4 = time at reception of answer
+
+    {[T3,NI,SI,T2,T1,NO,SO,Sz], Bytes} = unmark(Bin),
+
+    Sz = size(Bin) - Bytes,  %% assert
+
+    {T1,
+     {NO, SO, diameter_lib:micro_diff(T2, T1)},  %% Outbound
+     {NI, SI, diameter_lib:micro_diff(T4, T3)},  %% Inbound
+     T4}.
+
+%% mark/2
+
+mark(Bin, T) ->
+    Info = term_to_binary([diameter_lib:now() | T]),
+    <<Info/binary, Bin/binary>>.
+    
 %% unmark/1
 
 unmark(Bin) ->
-    binary_to_term(Bin).
+    T = binary_to_term(Bin),
+    {T, size(term_to_binary(T))}.
+
+%% ===========================================================================
+
+%% send_many_from_one/0
+%%
+%% Demonstrates sluggish delivery of messages.
+
+send_many_from_one() ->
+    [{timetrap, {seconds, 30}}].
+
+send_many_from_one(_) ->
+    ?OK = send_multiple(1, 128, 1024).
 
 %% ===========================================================================
 
@@ -345,7 +419,7 @@ unmark(Bin) ->
 %% Demonstrates reception of a message that differs from that sent.
 
 receive_what_was_sent(_Config) ->
-    send_from_multiple_clients(1, 1024*32).  %% fails
+    ?OK = send_multiple(1, 1, 1024*32).
 
 %% ===========================================================================
 
@@ -357,16 +431,23 @@ open() ->
 %% open/1
 
 open(Opts) ->
-    gen_sctp:open([{ip, ?ADDR}, {port, 0}, {active, true}, binary,
+    gen_sctp:open([{ip, ?ADDR}, {port, 0}, {active, false}, binary,
+                   {sctp_initmsg, #sctp_initmsg{num_ostreams = ?STREAMS,
+                                                max_instreams = ?STREAMS}},
                    {recbuf, 1 bsl 16}, {sndbuf, 1 bsl 16}
                    | Opts]).
 
-%% assoc/1
+%% report/2
 
-assoc(Sock) ->
-    receive
-        ?SCTP(Sock, {_, #sctp_assoc_change{state = S,
-                                           assoc_id = Id}}) ->
-            comm_up = S,  %% assert
-            Id
-    end.
+report(T0, Ts) ->
+    ct:pal("~p~n", [lists:sort([sort([{diameter_lib:micro_diff(T1,T0),
+                                       OT,
+                                       IT,
+                                       diameter_lib:micro_diff(T4,T0)}
+                                      || {T1,OT,IT,T4} <- L])
+                                || L <- Ts])]).
+
+%% sort/1
+
+sort(L) ->
+    lists:sort(fun({_,{N,_,_},_,_}, {_,{M,_,_},_,_}) -> N =< M end, L).
diff --git a/lib/diameter/test/diameter_gen_tcp_SUITE.erl b/lib/diameter/test/diameter_gen_tcp_SUITE.erl
index 2be2cf4b35..db42ea813e 100644
--- a/lib/diameter/test/diameter_gen_tcp_SUITE.erl
+++ b/lib/diameter/test/diameter_gen_tcp_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2014-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2014-2017. 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.
@@ -54,7 +54,7 @@ all() ->
 
 send_long(_) ->
     {Sock, SendF} = connection(),
-    B = list_to_binary(lists:duplicate(1 bsl 20, $X)),
+    B = binary:copy(<<$X>>, 1 bsl 20),
     ok = SendF(B),
     B = recv(Sock, size(B), []).
 
diff --git a/lib/diameter/test/diameter_traffic_SUITE.erl b/lib/diameter/test/diameter_traffic_SUITE.erl
index 4c82d4dee2..c6d63a9345 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-2016. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2017. 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.
@@ -27,6 +27,8 @@
 -export([suite/0,
          all/0,
          groups/0,
+         init_per_suite/1,
+         end_per_suite/1,
          init_per_group/2,
          end_per_group/2,
          init_per_testcase/2,
@@ -90,7 +92,6 @@
          send_multiple_filters_2/1,
          send_multiple_filters_3/1,
          send_anything/1,
-         outstanding/1,
          remove_transports/1,
          empty/1,
          stop_services/1,
@@ -106,6 +107,9 @@
          handle_error/6,
          handle_request/3]).
 
+%% diameter_{tcp,sctp} callbacks
+-export([message/3]).
+
 -include("diameter.hrl").
 -include("diameter_gen_base_rfc3588.hrl").
 -include("diameter_gen_base_accounting.hrl").
@@ -145,21 +149,29 @@
 
 %% Whether to decode stringish Diameter types to strings, or leave
 %% them as binary.
--define(STRING_DECODES, [true, false]).
+-define(STRING_DECODES, [false, true]).
 
 %% Which transport protocol to use.
 -define(TRANSPORTS, [tcp, sctp]).
 
+%% Send from a dedicated process?
+-define(SENDERS, [true, false]).
+
+%% Message callbacks from diameter_{tcp,sctp}?
+-define(CALLBACKS, [true, false]).
+
 -record(group,
         {transport,
+         strings,
          client_service,
          client_encoding,
          client_dict0,
-         client_strings,
+         client_sender,
          server_service,
          server_encoding,
          server_container,
-         server_strings}).
+         server_sender,
+         server_throttle}).
 
 %% Not really what we should be setting unless the message is sent in
 %% the common application but diameter doesn't care.
@@ -243,76 +255,128 @@ suite() ->
     [{timetrap, {seconds, 10}}].
 
 all() ->
-    [start, result_codes, {group, traffic}, outstanding, empty, stop].
+    [start, result_codes, {group, traffic}, empty, stop].
 
 groups() ->
-    Ts = tc(),
-    Sctp = ?util:have_sctp(),
-    [{B, [P], Ts} || {B,P} <- [{true, shuffle}, {false, parallel}]]
+    [{P, [P], Ts} || Ts <- [tc(tc())], P <- [shuffle, parallel]]
         ++
-        [{?util:name([T,R,D,A,C,SD,CD]),
+        [{?util:name([T,R,D,A,C,S,SS,ST,CS]),
           [],
-          [start_services,
-           add_transports,
-           result_codes,
-           {group, SD orelse CD},
-           remove_transports,
-           stop_services]}
+          [{group, if S -> shuffle; not S -> parallel end}]}
          || T <- ?TRANSPORTS,
-            T /= sctp orelse Sctp,
             R <- ?ENCODINGS,
             D <- ?RFCS,
             A <- ?ENCODINGS,
             C <- ?CONTAINERS,
-            SD <- ?STRING_DECODES,
-            CD <- ?STRING_DECODES]
+            S <- ?STRING_DECODES,
+            SS <- ?SENDERS,
+            ST <- ?CALLBACKS,
+            CS <- ?SENDERS]
+        ++
+        [{T, [], groups([[T,R,D,A,C,S,SS,ST,CS]
+                         || R <- ?ENCODINGS,
+                            D <- ?RFCS,
+                            A <- ?ENCODINGS,
+                            C <- ?CONTAINERS,
+                            S <- ?STRING_DECODES,
+                            SS <- ?SENDERS,
+                            ST <- ?CALLBACKS,
+                            CS <- ?SENDERS,
+                            SS orelse CS])}  %% avoid deadlock
+         || T <- ?TRANSPORTS]
         ++
-        [{traffic, [], [{group, ?util:name([T,R,D,A,C,SD,CD])}
-                        || T <- ?TRANSPORTS,
-                           T /= sctp orelse Sctp,
-                           R <- ?ENCODINGS,
-                           D <- ?RFCS,
-                           A <- ?ENCODINGS,
-                           C <- ?CONTAINERS,
-                           SD <- ?STRING_DECODES,
-                           CD <- ?STRING_DECODES]}].
+        [{traffic, [], [{group, T} || T <- ?TRANSPORTS]}].
+
+%groups(_) ->  %% debug
+%    Name = [sctp,record,rfc6733,record,pkt,false,false,false,false],
+%    [{group, ?util:name(Name)}];
+groups(Names) ->
+    [{group, ?util:name(L)} || L <- Names].
+
+%tc([N|_]) ->  %% debug
+%    [N];
+tc(L) ->
+    L.
+
+%% --------------------
+
+init_per_suite(Config) ->
+    [{sctp, ?util:have_sctp()} | Config].
+
+end_per_suite(_Config) ->
+    ok.
+
+%% --------------------
+
+init_per_group(Name, Config)
+  when Name == shuffle;
+       Name == parallel ->
+    start_services(Config),
+    add_transports(Config);
+
+init_per_group(sctp = Name, Config) ->
+    {_, Sctp} = lists:keyfind(Name, 1, Config),
+    if Sctp ->
+            Config;
+       true ->
+            {skip, Name}
+    end;
 
 init_per_group(Name, Config) ->
     case ?util:name(Name) of
-        [T,R,D,A,C,SD,CD] ->
+        [T,R,D,A,C,S,SS,ST,CS] ->
             G = #group{transport = T,
+                       strings = S,
                        client_service = [$C|?util:unique_string()],
                        client_encoding = R,
                        client_dict0 = dict0(D),
-                       client_strings = CD,
+                       client_sender = CS,
                        server_service = [$S|?util:unique_string()],
                        server_encoding = A,
                        server_container = C,
-                       server_strings = SD},
-            [{group, G} | Config];
+                       server_sender = SS,
+                       server_throttle = ST},
+            %% Limit the number of testcase, since the number of
+            %% groups is large.
+            All = ?util:scramble(tc()),
+            TCs = lists:sublist(All, rand:uniform(32)),
+            [{group, G}, {runlist, TCs} | Config];
         _ ->
             Config
     end.
 
+end_per_group(Name, Config)
+  when Name == shuffle;
+       Name == parallel ->
+    remove_transports(Config),
+    stop_services(Config);
+
 end_per_group(_, _) ->
     ok.
 
+%% --------------------
+
 %% Skip testcases that can reasonably fail under SCTP.
 init_per_testcase(Name, Config) ->
-    case [skip || #group{transport = sctp}
-                      <- [proplists:get_value(group, Config)],
-                  send_maxlen == Name
-                      orelse send_long == Name]
+    TCs = proplists:get_value(runlist, Config, []),
+    Run = [] == TCs orelse lists:member(Name, TCs),
+    case [G || #group{transport = sctp} = G
+                   <- [proplists:get_value(group, Config)]]
     of
-        [skip] ->
+        [_] when Name == send_maxlen;
+                 Name == send_long ->
             {skip, sctp};
-        [] ->
+        _ when not Run ->
+            {skip, random};
+        _ ->
             [{testcase, Name} | Config]
     end.
 
 end_per_testcase(_, _) ->
     ok.
 
+%% --------------------
+
 %% Testcases to run when services are started and connections
 %% established.
 tc() ->
@@ -377,28 +441,34 @@ start(_Config) ->
     ok = diameter:start().
 
 start_services(Config) ->
-    #group{client_service = CN,
-           client_strings = CD,
-           server_service = SN,
-           server_strings = SD}
+    #group{strings = S,
+           client_service = CN,
+           server_service = SN}
         = group(Config),
-    ok = diameter:start_service(SN, ?SERVICE(SN, SD)),
+    ok = diameter:start_service(SN, ?SERVICE(SN, S)),
     ok = diameter:start_service(CN, [{sequence, ?CLIENT_MASK}
-                                     | ?SERVICE(CN, CD)]).
+                                     | ?SERVICE(CN, S)]).
 
 add_transports(Config) ->
     #group{transport = T,
            client_service = CN,
-           server_service = SN}
+           client_sender = CS,
+           server_service = SN,
+           server_sender = SS,
+           server_throttle = ST}
         = group(Config), 
     LRef = ?util:listen(SN,
-                        T,
+                        [T,
+                         {sender, SS},
+                         {message_cb, ST andalso {?MODULE, message, [4]}}
+                         | [{packet, hd(?util:scramble([false, raw]))}
+                            || T == sctp andalso CS]],
                         [{capabilities_cb, fun capx/2},
                          {pool_size, 8},
                          {spawn_opt, [{min_heap_size, 8096}]},
                          {applications, apps(rfc3588)}]),
     Cs = [?util:connect(CN,
-                        T,
+                        [T, {sender, CS}],
                         LRef,
                         [{id, Id},
                          {capabilities, [{'Origin-State-Id', origin(Id)}]},
@@ -415,11 +485,6 @@ apps(D0) ->
     D = dict0(D0),
     [acct(D), D].
 
-%% Ensure there are no outstanding requests in request table.
-outstanding(_Config) ->
-    [] = [T || T <- ets:tab2list(diameter_request),
-               is_atom(element(1,T))].
-
 remove_transports(Config) ->
     #group{client_service = CN,
            server_service = SN}
@@ -689,14 +754,14 @@ send_unexpected_mandatory(Config) ->
 %% Send something long that will be fragmented by TCP.
 send_long(Config) ->
     Req = ['STR', {'Termination-Cause', ?LOGOUT},
-                  {'User-Name', [lists:duplicate(1 bsl 20, $X)]}],
+                  {'User-Name', [binary:copy(<<$X>>, 1 bsl 20)]}],
     ['STA', {'Session-Id', _}, {'Result-Code', ?SUCCESS} | _]
         = call(Config, Req).
 
 %% Send something longer than the configure incoming_maxlen.
 send_maxlen(Config) ->
     Req = ['STR', {'Termination-Cause', ?LOGOUT},
-                  {'User-Name', [lists:duplicate(1 bsl 21, $X)]}],
+                  {'User-Name', [binary:copy(<<$X>>, 1 bsl 21)]}],
     {timeout, _} = call(Config, Req).
 
 %% Send something for which pick_peer finds no suitable peer.
@@ -875,7 +940,7 @@ group(Config) ->
     #group{} = proplists:get_value(group, Config).
 
 string(V, Config) ->
-    #group{client_strings = B} = group(Config),
+    #group{strings = B} = group(Config),
     decode(V,B).
 
 decode(S, true)
@@ -995,7 +1060,7 @@ pick_peer(Peers, _, [$C|_], _State, {send_detach, Group}, _, {_,_}) ->
 find(#group{client_service = CN,
             server_encoding = A,
             server_container = C},
-     Peers) ->
+     [_|_] = Peers) ->
     Id = {A,C},
     [P] = [P || P <- Peers, id(Id, P, CN)],
     {ok, P}.
@@ -1429,3 +1494,33 @@ request(#diameter_base_STR{'Session-Id' = SId},
 %% send_error/send_timeout
 request(#diameter_base_RAR{}, _Caps) ->
     receive after 2000 -> {protocol_error, ?TOO_BUSY} end.
+
+%% message/3
+%%
+%% Limit the number of messages received. More can be received if read
+%% in the same packet.
+
+message(recv = D, {[_], Bin}, N) ->
+    message(D, Bin, N);
+message(Dir, #diameter_packet{bin = Bin}, N) ->
+    message(Dir, Bin, N);
+
+%% incoming request
+message(recv, <<_:32, 1, _/bits>> = Bin, N) ->
+    [Bin, 1 < N, fun ?MODULE:message/3, N-1];
+
+%% incoming answer
+message(recv, Bin, _) ->
+    [Bin];
+
+%% outgoing
+message(send, Bin, _) ->
+    [Bin];
+
+%% sent request
+message(ack, <<_:32, 1, _/bits>>, _) ->
+    [];
+
+%% sent answer or discarded request
+message(ack, _, N) ->
+    [0 =< N, fun ?MODULE:message/3, N+1].
diff --git a/lib/diameter/test/diameter_transport_SUITE.erl b/lib/diameter/test/diameter_transport_SUITE.erl
index c94f46b7a5..9d981d0a2b 100644
--- a/lib/diameter/test/diameter_transport_SUITE.erl
+++ b/lib/diameter/test/diameter_transport_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2017. 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.
@@ -294,10 +294,17 @@ init(gen_accept, {Prot, Ref}) ->
     {ok, PortNr} = inet:port(LSock),
     true = diameter_reg:add_new(?TEST_LISTENER(Ref, PortNr)),
 
-    %% Accept a connection, receive a message and send it back.
+    %% Accept a connection, receive a message send it back, and wait
+    %% for the peer to close the connection.
     {ok, Sock} = gen_accept(Prot, LSock),
     Bin = gen_recv(Prot, Sock),
-    ok = gen_send(Prot, Sock, Bin);
+    ok = gen_send(Prot, Sock, Bin),
+    receive
+        {tcp_closed, Sock} = T ->
+            T;
+        ?SCTP(Sock, {_, #sctp_assoc_change{}}) = T ->
+            T
+    end;
 
 init(connect, {Prot, Ref}) ->
     %% Lookup the peer's listening socket.
@@ -311,12 +318,7 @@ init(connect, {Prot, Ref}) ->
     %% Send a message and receive it back.
     Bin = make_msg(),
     TPid ! ?TMSG({send, Bin}),
-    Bin = bin(Prot, ?RECV(?TMSG({recv, P}), P)),
-
-    %% Expect the transport process to die as a result of the peer
-    %% closing the connection.
-    MRef = erlang:monitor(process, TPid),
-    ?RECV({'DOWN', MRef, process, _, _}).
+    Bin = bin(Prot, ?RECV(?TMSG({recv, P}), P)).
 
 bin(sctp, #diameter_packet{bin = Bin}) ->
     Bin;
@@ -336,15 +338,11 @@ make_msg() ->
     <<1:8, Len:24, Bin/binary>>.
 
 %% crypto:rand_bytes/1 isn't available on all platforms (since openssl
-%% isn't) so roll our own.
+%% isn't) so roll our own. Not particularly random, but less verbose
+%% in trace.
 rand_bytes(N) ->
-    rand_bytes(N, <<>>).
-
-rand_bytes(0, Bin) ->
-    Bin;
-rand_bytes(N, Bin) ->
     Oct = rand:uniform(256) - 1,
-    rand_bytes(N-1, <<Oct, Bin/binary>>).
+    binary:copy(<<Oct>>, N).
 
 %% ===========================================================================
 
diff --git a/lib/diameter/test/diameter_util.erl b/lib/diameter/test/diameter_util.erl
index cca28dd23c..03f79096ac 100644
--- a/lib/diameter/test/diameter_util.erl
+++ b/lib/diameter/test/diameter_util.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2016. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2017. 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.
@@ -172,18 +172,7 @@ recvl([{MRef, F} | L], Ref, Fun, Acc) ->
 %% Sort a list into random order.
 
 scramble(L) ->
-    foldl(fun(true, _, S, false) -> S end,
-          false,
-          [[fun s/1, L]]).
-
-s(L) ->
-    s([], L).
-
-s(Acc, []) ->
-    Acc;
-s(Acc, L) ->
-    {H, [T|Rest]} = lists:split(rand:uniform(length(L)) - 1, L),
-    s([T|Acc], H ++ Rest).
+    [X || {_,X} <- lists:sort([{rand:uniform(), T} || T <- L])].
 
 %% ---------------------------------------------------------------------------
 %% unique_string/0
@@ -195,21 +184,22 @@ unique_string() ->
 %% have_sctp/0
 
 have_sctp() ->
-    case erlang:system_info(system_architecture) of
-	%% We do not support the sctp version present in solaris
-	%% version "sparc-sun-solaris2.10", that behaves differently
-	%% from later versions and linux
-	"sparc-sun-solaris2.10" ->
-	    false;
-	_->
-	    case gen_sctp:open() of
-		{ok, Sock} ->
-		    gen_sctp:close(Sock),
-		    true;
-		{error, E} when E == eprotonosupport;
-				E == esocktnosupport -> %% fail on any other reason
-		    false
-	    end
+    have_sctp(erlang:system_info(system_architecture)).
+
+%% Don't run SCTP on platforms where it's either known to be flakey or
+%% isn't available.
+
+have_sctp("sparc-sun-solaris2.10") ->
+    false;
+
+have_sctp(_) ->
+    case gen_sctp:open() of
+        {ok, Sock} ->
+            gen_sctp:close(Sock),
+            true;
+        {error, E} when E == eprotonosupport;
+                        E == esocktnosupport -> %% fail on any other reason
+            false
     end.
 
 %% ---------------------------------------------------------------------------
@@ -313,17 +303,23 @@ listen(SvcName, Prot, Opts) ->
 connect(Client, Prot, LRef) ->
     connect(Client, Prot, LRef, []).
 
-connect(Client, Prot, LRef, Opts) ->
+connect(Client, ProtOpts, LRef, Opts) ->
+    Prot = head(ProtOpts),
     [PortNr] = lport(Prot, LRef),
     Client = diameter:service_info(Client, name),  %% assert
     true = diameter:subscribe(Client),
-    Ref = add_transport(Client, {connect, opts(Prot, PortNr) ++ Opts}),
+    Ref = add_transport(Client, {connect, opts(ProtOpts, PortNr) ++ Opts}),
     true = transport(Client, Ref),                 %% assert
 
     diameter_lib:for_n(fun(_) -> ok = up(Client, Ref, Prot, PortNr) end,
                        proplists:get_value(pool_size, Opts, 1)),
     Ref.
 
+head([T|_]) ->
+    T;
+head(T) ->
+    T.
+
 up(Client, Ref, Prot, PortNr) ->
     receive
         {diameter_event, Client, {up, Ref, _, _, _}} -> ok
@@ -366,10 +362,13 @@ tmod(sctp) ->
 tmod(any) ->
     [diameter_sctp, diameter_tcp].
 
-opts(Prot, T) ->
-    tmo(T, lists:append([[{transport_module, M}, {transport_config, C}]
+opts([Prot | Opts], T) ->
+    tmo(T, lists:append([[{transport_module, M}, {transport_config, C ++ Opts}]
                          || M <- tmod(Prot),
-                            C <- [cfg(M,T) ++ cfg(M) ++ cfg(T)]])).
+                            C <- [cfg(M,T) ++ cfg(M) ++ cfg(T)]]));
+
+opts(Prot, T) ->
+    opts([Prot], T).
 
 tmo(listen, Opts) ->
     Opts;
diff --git a/lib/diameter/test/diameter_watchdog_SUITE.erl b/lib/diameter/test/diameter_watchdog_SUITE.erl
index 6d22ddcc18..39c4f051a5 100644
--- a/lib/diameter/test/diameter_watchdog_SUITE.erl
+++ b/lib/diameter/test/diameter_watchdog_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2015. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2017. 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,13 +44,8 @@
 -export([peer_up/3,
          peer_down/3]).
 
-%% gen_tcp-ish interface
--export([listen/2,
-         accept/1,
-         connect/3,
-         send/2,
-         setopts/2,
-         close/1]).
+%% diameter_tcp message_cb
+-export([message/3]).
 
 -include("diameter.hrl").
 -include("diameter_ct.hrl").
@@ -161,9 +156,9 @@ reopen(Type, Test, Ref, Wd, N, M) ->
     reopen(Type, Test, SvcName, TRef, Wd, N, M).
 
 cfg(Type, Type, Wd) ->
-    {Wd, [], []};
+    {Wd, [], false};
 cfg(_Type, _Test, _Wd) ->
-    {?WD(?PEER_WD), [{okay, 0}], [{module, ?MODULE}]}.
+    {?WD(?PEER_WD), [{okay, 0}], true}.
 
 %% reopen/7
 
@@ -346,7 +341,7 @@ recv_reopen(listen, Ref) ->
 %% reg/3
 %%
 %% Lookup the pid of the transport process and publish a term for
-%% send/2 to lookup.
+%% message/3 to lookup.
 reg(TRef, SvcName, T) ->
     TPid = tpid(TRef, diameter:service_info(SvcName, transport)),
     true = diameter_reg:add_new({?MODULE, TPid, T}).
@@ -394,7 +389,7 @@ suspect(_) ->
 suspect(Type, Fake, Ref, N)
   when is_reference(Ref) ->
     {SvcName, TRef}
-        = start(Type, Ref, {?WD(10000), [{suspect, N}], mod(Fake)}),
+        = start(Type, Ref, {?WD(10000), [{suspect, N}], Fake}),
     {initial, okay} = ?WD_EVENT(TRef),
     suspect(TRef, Fake, SvcName, N);
 
@@ -436,11 +431,6 @@ abuse([F|A], Test) ->
 abuse(F, Test) ->
     abuse([F], Test).
 
-mod(true) ->
-    [{module, ?MODULE}];
-mod(false) ->
-    [].
-
 %% ===========================================================================
 %% # okay/1
 %% ===========================================================================
@@ -456,7 +446,7 @@ okay(Type, Fake, Ref, N)
     {SvcName, TRef}
         = start(Type, Ref, {?WD(10000),
                             [{okay, choose(Fake, 0, N)}],
-                            mod(Fake)}),
+                            Fake}),
     {initial, okay} = ?WD_EVENT(TRef),
     okay(TRef,
          Fake,
@@ -515,12 +505,17 @@ start(Type, Ref, T) ->
     true = diameter_reg:add_new({Type, Ref, Name}),
     {Name, TRef}.
 
-opts(Type, Ref, {Timer, Config, Mod}) ->
+opts(Type, Ref, {Timer, Config, Fake})
+  when is_boolean(Fake) ->
     [{transport_module, diameter_tcp},
-     {transport_config, Mod ++ [{ip, ?ADDR}, {port, 0}] ++ cfg(Type, Ref)},
+     {transport_config, mod(Fake) ++ [{ip, ?ADDR}, {port, 0}]
+                                  ++ cfg(Type, Ref)},
      {watchdog_timer, Timer},
      {watchdog_config, Config}].
 
+mod(B) ->
+    [{message_cb, [fun message/3, capx]} || B].
+
 cfg(listen, _) ->
     [];
 cfg(connect, Ref) ->
@@ -531,37 +526,29 @@ cfg(connect, Ref) ->
 
 %% ===========================================================================
 
-listen(PortNr, Opts) ->
-    gen_tcp:listen(PortNr, Opts).
-
-accept(LSock) ->
-    gen_tcp:accept(LSock).
+%% message/3
 
-connect(Addr, Port, Opts) ->
-    gen_tcp:connect(Addr, Port, Opts).
+message(send, Bin, X) ->
+    send(Bin, X);
 
-setopts(Sock, Opts) ->
-    inet:setopts(Sock, Opts).
+message(recv, Bin, _) ->
+    [Bin];
 
-send(Sock, Bin) ->
-    send(getr(config), Sock, Bin).
-
-close(Sock) ->
-    gen_tcp:close(Sock).
+message(_, _, _) ->
+    [].
 
-%% send/3
+%% send/2
 
 %% First outgoing message from a new transport process is CER/CEA.
 %% Remaining outgoing messages are either DWR or DWA.
-send(undefined, Sock, Bin) ->
-    <<_:32, _:8, 257:24, _/binary>> = Bin,
-    putr(config, init),
-    gen_tcp:send(Sock, Bin);
+send(Bin, capx) ->
+    <<_:32, _:8, 257:24, _/binary>> = Bin,  %% assert on CER/CEA
+    [Bin, fun message/3, init];
 
 %% Outgoing DWR: fake reception of DWA. Use the fact that AVP values
 %% are ignored. This is to ensure that the peer's watchdog state
 %% transitions are only induced by responses to messages it sends.
-send(_, Sock, <<_:32, 1:1, _:7, 280:24, _:32, EId:32, HId:32, _/binary>>) ->
+send(<<_:32, 1:1, _:7, 280:24, _:32, EId:32, HId:32, _/binary>>, _) ->
     Pkt = #diameter_packet{header = #diameter_header{version = 1,
                                                      end_to_end_id = EId,
                                                      hop_by_hop_id = HId},
@@ -569,47 +556,36 @@ send(_, Sock, <<_:32, 1:1, _:7, 280:24, _:32, EId:32, HId:32, _/binary>>) ->
                                          {'Origin-Host', "XXX"},
                                          {'Origin-Realm', ?REALM}]},
     #diameter_packet{bin = Bin} = diameter_codec:encode(?BASE, Pkt),
-    self() ! {tcp, Sock, Bin},
-    ok;
+    [recv, Bin];
 
 %% First outgoing DWA.
-send(init, Sock, Bin) ->
+send(Bin, init) ->
     [{{?MODULE, _, T}, _}] = diameter_reg:wait({?MODULE, self(), '_'}),
-    putr(config, T),
-    send(Sock, Bin);
+    send(Bin, T);
 
 %% First transport process.
-send({SvcName, {_,_,_} = T}, Sock, Bin) ->
+send(Bin, {SvcName, {_,_,_} = T}) ->
     [{'Origin-Host', _} = OH, {'Origin-Realm', _} = OR | _]
         = ?SERVICE(SvcName),
     putr(origin, [OH, OR]),
-    putr(config, T),
-    send(Sock, Bin);
+    send(Bin, T);
 
 %% Discard DWA, failback after another timeout in the peer.
-send({Wd, 0 = No, Msg}, Sock, Bin) ->
+send(Bin, {Wd, 0 = No, Msg}) ->
     Origin = getr(origin),
-    spawn(fun() -> failback(?ONE_WD(Wd), Msg, Sock, Bin, Origin) end),
-    putr(config, No),
-    ok;
+    [{defer, ?ONE_WD(Wd), [msg(Msg, Bin, Origin)]}, fun message/3, No];
 
 %% Send DWA while we're in the mood (aka 0 < N).
-send({Wd, N, Msg}, Sock, Bin) ->
-    putr(config, {Wd, N-1, Msg}),
-    gen_tcp:send(Sock, Bin);
+send(Bin, {Wd, N, Msg}) ->
+    [Bin, fun message/3, {Wd, N-1, Msg}];
 
 %% Discard DWA.
-send(0, _Sock, _Bin) ->
-    ok;
+send(_Bin, 0 = No) ->
+    [fun message/3, No];
 
 %% Send DWA.
-send(N, Sock, <<_:32, 0:1, _:7, 280:24, _/binary>> = Bin) ->
-    putr(config, N-1),
-    gen_tcp:send(Sock, Bin).
-
-failback(Tmo, Msg, Sock, Bin, Origin) ->
-    timer:sleep(Tmo),
-    ok = gen_tcp:send(Sock, msg(Msg, Bin, Origin)).
+send(<<_:32, 0:1, _:7, 280:24, _/binary>> = DWA, N) ->
+    [DWA, fun message/3, N-1].
 
 %% msg/2