channel_source_SUITE: Wait for channel termination

... which could take some time after connection close. This is mostly a
guess, based on the transient failures I see in CI. Hopefully it will
help there.

Author: dumbbell

test/channel_source_SUITE.erl

@@ -17,6 +17,7 @@
 -module(channel_source_SUITE).
 
 -include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("eunit/include/eunit.hrl").
 
 -compile(export_all).
 
@@ -85,7 +86,7 @@ network_rabbit_reader_channel_source1(Config) ->
     [{source, ?MODULE}] = rabbit_channel:info(ServerCh, [source]),
     amqp_channel:close(ClientCh),
     amqp_connection:close(Conn),
-    {error, channel_terminated} = rabbit_channel:source(ServerCh, ?MODULE),
+    wait_for_channel_termination(ServerCh, 60),
     passed.
 
 network_arbitrary_channel_source(Config) ->
@@ -108,7 +109,7 @@ network_arbitrary_channel_source1(Config) ->
     [{source, ?MODULE}] = rabbit_channel:info(Ch, [source]),
     [exit(P, normal) || P <- [Writer, Limiter, Collector, Ch]],
     amqp_connection:close(Conn),
-    {error, channel_terminated} = rabbit_channel:source(Ch, ?MODULE),
+    wait_for_channel_termination(Ch, 60),
     passed.
 
 direct_channel_source(Config) ->
@@ -125,9 +126,22 @@ direct_channel_source1(Config) ->
     [{source, ?MODULE}] = rabbit_channel:info(ServerCh, [source]),
     amqp_channel:close(ClientCh),
     amqp_connection:close(Conn),
-    {error, channel_terminated} = rabbit_channel:source(ServerCh, ?MODULE),
+    wait_for_channel_termination(ServerCh, 60),
     passed.
 
+wait_for_channel_termination(Ch, 0) ->
+    ?assertEqual(
+       {error, channel_terminated},
+       rabbit_channel:source(Ch, ?MODULE));
+wait_for_channel_termination(Ch, Attempts) ->
+    case rabbit_channel:source(Ch, ?MODULE) of
+        {error, channel_terminated} ->
+            ok;
+        _ ->
+            timer:sleep(1000),
+            wait_for_channel_termination(Ch, Attempts - 1)
+    end.
+
 undefined_channel_source(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
       ?MODULE, undefined_channel_source1, [Config]).
@@ -142,7 +156,7 @@ undefined_channel_source1(_Config) ->
     passed.
 
 wait_for_server_channel(ExistingChannels, ServerCh, 0) ->
-    [ServerCh] = rabbit_channel:list() -- ExistingChannels;
+    ?assertEqual([ServerCh], rabbit_channel:list() -- ExistingChannels);
 wait_for_server_channel(ExistingChannels, ServerCh, Attempts) ->
     case rabbit_channel:list() -- ExistingChannels of
         [ServerCh] ->