Merge pull request #2447 from rabbitmq/qq-peek-command

Quorum queue peek

2 files changed, 47 insertions, 1 deletions

diff --git a/test/quorum_queue_SUITE.erl b/test/quorum_queue_SUITE.erl
index 15f3558046..942b42f5e4 100644
--- a/test/quorum_queue_SUITE.erl
+++ b/test/quorum_queue_SUITE.erl
@@ -131,7 +131,8 @@ all_tests() ->
      invalid_policy,
      delete_if_empty,
      delete_if_unused,
-     queue_ttl
+     queue_ttl,
+     peek
     ].
 
 memory_tests() ->
@@ -2397,6 +2398,39 @@ queue_length_in_memory_purge(Config) ->
     ?assertEqual([{0, 0}],
                  dirty_query([Server], RaName, fun rabbit_fifo:query_in_memory_usage/1)).
 
+peek(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                                  {<<"x-max-in-memory-length">>, long, 2}])),
+
+    Msg1 = <<"msg1">>,
+    Msg2 = <<"msg11">>,
+
+    QName = rabbit_misc:r(<<"/">>, queue, QQ),
+    ?assertMatch({error, no_message_at_pos},
+                 rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+                                              peek, [1, QName])),
+    publish(Ch, QQ, Msg1),
+    publish(Ch, QQ, Msg2),
+    wait_for_messages(Config, [[QQ, <<"2">>, <<"2">>, <<"0">>]]),
+
+    ?assertMatch({ok, [_|_]},
+                 rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+                                              peek, [1, QName])),
+    ?assertMatch({ok, [_|_]},
+                 rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+                                              peek, [2, QName])),
+    ?assertMatch({error, no_message_at_pos},
+                 rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue,
+                                              peek, [3, QName])),
+
+    wait_for_messages(Config, [[QQ, <<"2">>, <<"2">>, <<"0">>]]),
+    ok.
+
 in_memory(Config) ->
     [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
diff --git a/test/rabbit_fifo_SUITE.erl b/test/rabbit_fifo_SUITE.erl
index 7778e04afb..8431dd8db7 100644
--- a/test/rabbit_fifo_SUITE.erl
+++ b/test/rabbit_fifo_SUITE.erl
@@ -1619,6 +1619,18 @@ queue_ttl_with_single_active_consumer_test(_) ->
 
     ok.
 
+query_peek_test(_) ->
+    State0 = test_init(test),
+    ?assertEqual({error, no_message_at_pos}, rabbit_fifo:query_peek(1, State0)),
+    {State1, _} = enq(1, 1, first, State0),
+    {State2, _} = enq(2, 2, second, State1),
+    ?assertMatch({ok, {_, {_, first}}}, rabbit_fifo:query_peek(1, State1)),
+    ?assertEqual({error, no_message_at_pos}, rabbit_fifo:query_peek(2, State1)),
+    ?assertMatch({ok, {_, {_, first}}}, rabbit_fifo:query_peek(1, State2)),
+    ?assertMatch({ok, {_, {_, second}}}, rabbit_fifo:query_peek(2, State2)),
+    ?assertEqual({error, no_message_at_pos}, rabbit_fifo:query_peek(3, State2)),
+    ok.
+
 %% Utility
 
 init(Conf) -> rabbit_fifo:init(Conf).