Support default exchange in consumer and publisher modules

lib/binding.ex

@@ -4,11 +4,13 @@ defmodule GenRMQ.Binding do
   declaring consumer bindings and exchanges.
   """
 
-  @type exchange :: String.t() | {exchange_kind(), String.t()}
+  @type exchange :: String.t() | {exchange_kind(), String.t()} | :default
   @type exchange_kind :: :topic | :direct | :fanout
 
   use AMQP
 
+  @default_exchange ""
+
   @doc false
   def bind_exchange_and_queue(chan, exchange, queue, routing_key) do
     declare_exchange(chan, exchange)
@@ -29,6 +31,8 @@ defmodule GenRMQ.Binding do
   end
 
   @doc false
+  def declare_exchange(_chan, :default), do: :ok
+
   def declare_exchange(chan, {:direct, exchange}) do
     Exchange.direct(chan, exchange, durable: true)
   end
@@ -45,6 +49,10 @@ defmodule GenRMQ.Binding do
     Exchange.topic(chan, exchange, durable: true)
   end
 
+  def exchange_name(:default) do
+    @default_exchange
+  end
+
   def exchange_name({_, exchange}) do
     exchange
   end
@@ -56,13 +64,15 @@ defmodule GenRMQ.Binding do
 
   defp bind_queue(chan, queue, exchange_name, routing_key) when is_list(routing_key) do
     Enum.reduce_while(routing_key, :ok, fn rk, acc ->
-      case Queue.bind(chan, queue, exchange_name, routing_key: rk) do
+      case bind_queue(chan, queue, exchange_name, rk) do
         :ok -> {:cont, acc}
         err -> {:halt, err}
       end
     end)
   end
 
+  defp bind_queue(_chan, _queue, :default, _routing_key), do: :ok
+
   defp bind_queue(chan, queue, exchange_name, routing_key) do
     Queue.bind(chan, queue, exchange_name, routing_key: routing_key)
   end

lib/consumer.ex

@@ -38,8 +38,8 @@ defmodule GenRMQ.Consumer do
 
   `queue` - the name of the queue to consume. If it does not exist, it will be created.
 
-  `exchange` - name or `{type, name}` of the exchange to which `queue` should be bound. If it does not exist, it will be created.
-  For valid exchange types see `GenRMQ.Binding`.
+  `exchange` - name, `:default` or `{type, name}` of the target exchange.
+  If it does not exist, it will be created. Supported types: `:direct`, `:fanout`, `:topic`
 
   `routing_key` - queue binding key, can also be a list.
 

lib/publisher.ex

@@ -42,8 +42,8 @@ defmodule GenRMQ.Publisher do
 
   `connection` - RabbitMQ connection options. Accepts same arguments as AMQP-library's [Connection.open/2](https://hexdocs.pm/amqp/AMQP.Connection.html#open/2).
 
-  `exchange` - name or `{type, name}` of the target exchange. If it does not exist, it will be created.
-  For valid exchange types see `GenRMQ.Binding`.
+  `exchange` - name, `:default` or `{type, name}` of the target exchange.
+  If it does not exist, it will be created. Supported types: `:direct`, `:fanout`, `:topic`
 
   ### Optional:
 

test/gen_rmq_consumer_test.exs

@@ -24,7 +24,8 @@ defmodule GenRMQ.ConsumerTest do
     WithTopicExchange,
     WithoutConcurrency,
     WithoutDeadletter,
-    WithoutReconnection
+    WithoutReconnection,
+    WithDefaultExchange
   }
 
   @connection "amqp://guest:guest@localhost:5672"
@@ -277,7 +278,7 @@ defmodule GenRMQ.ConsumerTest do
       with_test_consumer(ErrorWithoutConcurrency)
     end
 
-    test "should receive invoke the handle_error callback if an error is encountered with no concurrency",
+    test "should invoke the handle_error callback if an error is encountered with no concurrency",
          %{consumer: consumer_pid, state: state} = context do
       clear_mailbox()
 
@@ -593,6 +594,29 @@ defmodule GenRMQ.ConsumerTest do
     close_connection_and_channels_after_shutdown_test()
   end
 
+  describe "TestConsumer.WithDefaultExchange" do
+    setup do
+      Agent.start_link(fn -> MapSet.new() end, name: WithDefaultExchange)
+      {:ok, context} = with_test_consumer(WithDefaultExchange)
+      routing_key = context.state.config[:queue][:name]
+      {:ok, Map.put(context, :routing_key, routing_key)}
+    end
+
+    receive_message_test(WithDefaultExchange)
+
+    reject_message_test()
+
+    reconnect_after_connection_failure_test(WithDefaultExchange)
+
+    terminate_after_queue_deletion_test()
+
+    exit_signal_after_queue_deletion_test()
+
+    close_connection_and_channels_after_deletion_test()
+
+    close_connection_and_channels_after_shutdown_test()
+  end
+
   describe "Telemetry events" do
     setup :attach_telemetry_handlers
 
@@ -654,7 +678,7 @@ defmodule GenRMQ.ConsumerTest do
     exchange = state.config[:exchange]
 
     on_exit(fn -> Process.exit(consumer_pid, :normal) end)
-    {:ok, %{consumer: consumer_pid, exchange: exchange, state: state}}
+    {:ok, %{consumer: consumer_pid, exchange: exchange, state: state, routing_key: "#"}}
   end
 
   defp clear_mailbox do

test/gen_rmq_publisher_test.exs

@@ -8,7 +8,8 @@ defmodule GenRMQ.PublisherTest do
   alias TestPublisher.{
     Default,
     RedeclaringExistingExchange,
-    WithConfirmations
+    WithConfirmations,
+    WithDefaultExchange
   }
 
   @connection "amqp://guest:guest@localhost:5672"
@@ -334,6 +335,24 @@ defmodule GenRMQ.PublisherTest do
     end
   end
 
+  describe "TestPublisher.WithDefaultExchange" do
+    setup do
+      with_test_publisher(WithDefaultExchange)
+    end
+
+    test "should publish a message to the default exchange", %{publisher: publisher_pid} = context do
+      message = %{"msg" => "with default exchange"}
+      # https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchange-default
+      # all queues are bound to the default exchange with a queue name as a binding key
+      routing_key = context.out_queue
+      publish_result = GenRMQ.Publisher.publish(publisher_pid, Jason.encode!(message), routing_key)
+
+      Assert.repeatedly(fn -> assert out_queue_count(context) >= 1 end)
+      assert match?({:ok, ^message, _}, get_message_from_queue(context))
+      assert {:ok, :confirmed} == publish_result
+    end
+  end
+
   describe "Telemetry events" do
     setup :attach_telemetry_handlers
 

test/support/assert.ex

@@ -10,7 +10,7 @@ defmodule GenRMQ.Test.Assert do
     Agent.start_link(fn -> Map.new() end, name: __MODULE__)
   end
 
-  def repeatedly(function, time \\ 10_000, interval \\ 250) do
+  def repeatedly(function, time \\ 15_000, interval \\ 250) do
     task = Task.async(fn -> repeatedly_loop(function, interval) end)
 
     case Task.yield(task, time) do

test/support/consumer_shared_tests.ex

@@ -8,7 +8,7 @@ defmodule ConsumerSharedTests do
       test "should receive a message", context do
         message = %{"msg" => "some message"}
 
-        publish_message(context[:rabbit_conn], context[:exchange], Jason.encode!(message))
+        publish_message(context[:rabbit_conn], context[:exchange], Jason.encode!(message), context[:routing_key])
 
         GenRMQ.Test.Assert.repeatedly(fn ->
           assert Agent.get(unquote(mod), fn set -> message in set end) == true
@@ -22,7 +22,7 @@ defmodule ConsumerSharedTests do
       test "should reject a message", %{state: state} = context do
         message = "reject"
 
-        publish_message(context[:rabbit_conn], context[:exchange], Jason.encode!(message))
+        publish_message(context[:rabbit_conn], context[:exchange], Jason.encode!(message), context[:routing_key])
 
         GenRMQ.Test.Assert.repeatedly(fn ->
           assert queue_count(context[:rabbit_conn], state.config[:queue][:dead_letter][:name]) == {:ok, 1}
@@ -37,7 +37,7 @@ defmodule ConsumerSharedTests do
         message = "disconnect"
         AMQP.Connection.close(state.conn)
 
-        publish_message(context[:rabbit_conn], context[:exchange], Jason.encode!(message))
+        publish_message(context[:rabbit_conn], context[:exchange], Jason.encode!(message), context[:routing_key])
 
         GenRMQ.Test.Assert.repeatedly(fn ->
           assert Agent.get(unquote(mod), fn set -> message in set end) == true

test/support/test_consumers.ex

@@ -446,6 +446,39 @@ defmodule TestConsumer do
     end
   end
 
+  defmodule WithDefaultExchange do
+    @moduledoc false
+    @behaviour GenRMQ.Consumer
+
+    def init() do
+      [
+        queue: "gen_rmq_with_default_exchange",
+        exchange: :default,
+        prefetch_count: "10",
+        connection: "amqp://guest:guest@localhost:5672",
+        queue_ttl: 1_000
+      ]
+    end
+
+    def consumer_tag() do
+      "TestConsumer.WithDefaultExchange"
+    end
+
+    def handle_message(%GenRMQ.Message{payload: "\"reject\""} = message) do
+      GenRMQ.Consumer.reject(message)
+    end
+
+    def handle_message(message) do
+      payload = Jason.decode!(message.payload)
+      Agent.update(__MODULE__, &MapSet.put(&1, payload))
+      GenRMQ.Consumer.ack(message)
+    end
+
+    def handle_error(message, _reason) do
+      GenRMQ.Consumer.reject(message)
+    end
+  end
+
   defmodule RedeclaringExistingExchange do
     @moduledoc false
     @behaviour GenRMQ.Consumer

test/support/test_publishers.ex

@@ -40,4 +40,18 @@ defmodule TestPublisher do
       ]
     end
   end
+
+  defmodule WithDefaultExchange do
+    @moduledoc false
+    @behaviour GenRMQ.Publisher
+
+    def init() do
+      [
+        exchange: :default,
+        connection: "amqp://guest:guest@localhost:5672",
+        app_id: :my_app_id,
+        enable_confirmations: true
+      ]
+    end
+  end
 end