test: test message TTL override

Signed-off-by: Matteo Cafasso <[email protected]>

Author: noxdafox

test/exchange_SUITE.erl

@@ -101,36 +101,43 @@ deduplicate_message(Config) ->
     publish_message(Channel, <<"test">>, "deduplicate-this"),
     publish_message(Channel, <<"test">>, "deduplicate-this"),
 
-    timer:sleep(1000),
-
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
     #'basic.get_empty'{} = amqp_channel:call(Channel, Get),
 
     %% Deduplication header absent
     publish_message(Channel, <<"test">>),
     publish_message(Channel, <<"test">>),
 
-    timer:sleep(1000),
-
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
 
 deduplicate_message_ttl(Config) ->
+    Get = #'basic.get'{queue = <<"test">>},
     Channel = rabbit_ct_client_helpers:open_channel(Config),
 
     #'exchange.declare_ok'{} = amqp_channel:call(
                                  Channel, make_exchange(<<"test">>, 10, 1000)),
     bind_new_queue(Channel, <<"test">>, <<"test">>),
 
+    %% Exchange default TTL
     publish_message(Channel, <<"test">>, "deduplicate-this"),
     timer:sleep(2000),
     publish_message(Channel, <<"test">>, "deduplicate-this"),
 
-    Get = #'basic.get'{queue = <<"test">>},
+    {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
+    {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
+
+    %% Message TTL override
+    Headers = [{<<"x-cache-ttl">>, long, 500}],
+    publish_message(Channel, <<"test">>, "deduplicate-that", Headers),
+    timer:sleep(800),
+    publish_message(Channel, <<"test">>, "deduplicate-that", Headers),
+
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
 
 deduplicate_message_cache_overflow(Config) ->
+    Get = #'basic.get'{queue = <<"test">>},
     Channel = rabbit_ct_client_helpers:open_channel(Config),
 
     #'exchange.declare_ok'{} = amqp_channel:call(
@@ -141,9 +148,6 @@ deduplicate_message_cache_overflow(Config) ->
     publish_message(Channel, <<"test">>, "deduplicate-that"),
     publish_message(Channel, <<"test">>, "deduplicate-this"),
 
-    timer:sleep(1000),
-
-    Get = #'basic.get'{queue = <<"test">>},
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get),
     {#'basic.get_ok'{}, _} = amqp_channel:call(Channel, Get).
@@ -172,7 +176,11 @@ publish_message(Ch, Ex) ->
     amqp_channel:cast(Ch, Publish, Msg).
 
 publish_message(Ch, Ex, D) ->
+    publish_message(Ch, Ex, D, []).
+
+publish_message(Ch, Ex, D, H) ->
+    Headers = [{<<"x-deduplication-header">>, longstr, D}] ++ H,
     Publish = #'basic.publish'{exchange = Ex, routing_key = <<"#">>},
-    Props = #'P_basic'{headers = [{<<"x-deduplication-header">>, longstr, D}]},
+    Props = #'P_basic'{headers = Headers},
     Msg = #amqp_msg{props = Props, payload = <<"payload">>},
     amqp_channel:cast(Ch, Publish, Msg).