Shovel: use lqueue over queue

the `queue' module is not aware of the queue
length, which means its `len/1` function is
O(n) while with `lqueue' it is O(1).

We rely on `queue:len/1' for a shovel metric,
so why not use a more efficient drop-in replacement
our team has developed in 2011.

(cherry picked from commit 7aa7433)

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_amqp091_shovel.erl

@@ -362,16 +362,16 @@ status(_) ->
     running.
 
 pending_count(#{dest := Dest}) ->
-    Pending = maps:get(pending, Dest, queue:new()),
-    queue:len(Pending).
+    Pending = maps:get(pending, Dest, lqueue:new()),
+    lqueue:len(Pending).
 
 add_pending(Elem, State = #{dest := Dest}) ->
-    Pending = maps:get(pending, Dest, queue:new()),
-    State#{dest => Dest#{pending => queue:in(Elem, Pending)}}.
+    Pending = maps:get(pending, Dest, lqueue:new()),
+    State#{dest => Dest#{pending => lqueue:in(Elem, Pending)}}.
 
 pop_pending(State = #{dest := Dest}) ->
-    Pending = maps:get(pending, Dest, queue:new()),
-    case queue:out(Pending) of
+    Pending = maps:get(pending, Dest, lqueue:new()),
+    case lqueue:out(Pending) of
         {empty, _} ->
             empty;
         {{value, Elem}, Pending2} ->

deps/rabbitmq_shovel/src/rabbit_local_shovel.erl

@@ -476,7 +476,7 @@ status(State) ->
     end.
 
 pending_count(#{dest := #{pending_delivery := Pending}}) ->
-    queue:len(Pending);
+    lqueue:len(Pending);
 pending_count(_) ->
     0.
 
@@ -947,12 +947,12 @@ is_blocked(_) ->
     false.
 
 add_pending_delivery(Elem, State = #{dest := Dest}) ->
-    Pending = maps:get(pending_delivery, Dest, queue:new()),
-    State#{dest => Dest#{pending_delivery => queue:in(Elem, Pending)}}.
+    Pending = maps:get(pending_delivery, Dest, lqueue:new()),
+    State#{dest => Dest#{pending_delivery => lqueue:in(Elem, Pending)}}.
 
 pop_pending_delivery(State = #{dest := Dest}) ->
-    Pending = maps:get(pending_delivery, Dest, queue:new()),
-    case queue:out(Pending) of
+    Pending = maps:get(pending_delivery, Dest, lqueue:new()),
+    case lqueue:out(Pending) of
         {empty, _} ->
             empty;
         {{value, Elem}, Pending2} ->

deps/rabbitmq_shovel/test/pending_count_SUITE.erl

@@ -69,13 +69,13 @@ amqp091_pending_count_empty_queue(_Config) ->
 
 amqp091_pending_count_with_messages(_Config) ->
     %% Test that pending_count returns correct count when messages are pending
-    PendingQueue = queue:from_list([{1, msg1}, {2, msg2}, {3, msg3}]),
+    PendingQueue = lqueue:from_list([{1, msg1}, {2, msg2}, {3, msg3}]),
     State = #{dest => #{pending => PendingQueue}},
     ?assertEqual(3, rabbit_amqp091_shovel:pending_count(State)).
 
 amqp091_pending_count_after_drain(_Config) ->
     %% Test that pending_count returns 0 after messages are drained
-    EmptyQueue = queue:new(),
+    EmptyQueue = lqueue:new(),
     State = #{dest => #{pending => EmptyQueue}},
     ?assertEqual(0, rabbit_amqp091_shovel:pending_count(State)).