Allow prefetch_count when consuming a channel.

lib/clients/rabbitmq.ex

@@ -11,7 +11,15 @@ defmodule ExRabbitPool.RabbitMQ do
   end
 
   @impl true
-  def consume(%Channel{} = channel, queue, consumer_pid \\ nil, options \\ []) do
+  def consume(channel, queue, consumer_pid \\ nil, options \\ [])
+
+  def consume(%Channel{} = channel, queue, consumer_pid, [prefetch_count: prefetch_count] = options) do
+    Logger.warn("[ExRabbitPool.RabbitMQ.consume] queue: #{inspect queue} setting prefetch_count to #{prefetch_count}")
+    :ok = Basic.qos(channel, prefetch_count: prefetch_count)
+    Basic.consume(channel, queue, consumer_pid, options)
+  end
+
+  def consume(%Channel{} = channel, queue, consumer_pid, options) do
     Basic.consume(channel, queue, consumer_pid, options)
   end
 

lib/consumer.ex

@@ -166,7 +166,7 @@ defmodule ExRabbitPool.Consumer do
       # process and monitors it handle crashes and reconnections
       defp handle_channel_checkout(
              {:ok, %{pid: channel_pid} = channel},
-             %{config: config, queue: queue, adapter: adapter, config: config} = state
+             %{config: config, queue: queue, adapter: adapter} = state
            ) do
         config = Keyword.get(config, :options, [])
 

test/integration/consumer_test.exs

@@ -1,6 +1,7 @@
 defmodule ExRabbitPool.ConsumerTest do
   use ExUnit.Case, async: false
 
+  import ExUnit.CaptureLog
   alias ExRabbitPool.Worker.SetupQueue
   alias ExRabbitPool.RabbitMQ
   alias AMQP.Queue
@@ -86,15 +87,19 @@ defmodule ExRabbitPool.ConsumerTest do
   end
 
   test "should be able to consume messages out of rabbitmq", %{pool_id: pool_id, queue: queue} do
-    pid = start_supervised!({TestConsumer, pool_id: pool_id, queue: queue})
-    :erlang.trace(pid, true, [:receive])
-
-    ExRabbitPool.with_channel(pool_id, fn {:ok, channel} ->
-      assert :ok = RabbitMQ.publish(channel, "#{queue}_exchange", "", "Hello Consumer!")
-      assert_receive {:trace, ^pid, :receive, {:basic_deliver, "Hello Consumer!", _}}, 1000
-      {:ok, result} = Queue.status(channel, queue)
-      assert result == %{consumer_count: 1, message_count: 0, queue: queue}
-    end)
+    logs =
+      capture_log(fn ->
+        pid = start_supervised!({TestConsumer, pool_id: pool_id, queue: queue, options: [prefetch_count: 19]})
+        :erlang.trace(pid, true, [:receive])
+
+        ExRabbitPool.with_channel(pool_id, fn {:ok, channel} ->
+          assert :ok = RabbitMQ.publish(channel, "#{queue}_exchange", "", "Hello Consumer!")
+          assert_receive {:trace, ^pid, :receive, {:basic_deliver, "Hello Consumer!", _}}, 1000
+          {:ok, result} = Queue.status(channel, queue)
+          assert result == %{consumer_count: 1, message_count: 0, queue: queue}
+        end)
+      end)
+    assert logs =~ "[ExRabbitPool.RabbitMQ.consume] queue: #{inspect queue} setting prefetch_count to 19"
   end
 
   test "should be able to consume messages out of rabbitmq with default consumer", %{