Add option to keep routing key

lib/message_queue/adapters/rabbitmq/producer_worker.ex

@@ -148,7 +148,7 @@ defmodule MessageQueue.Adapters.RabbitMQ.ProducerWorker do
   end
 
   defp get_exchange("" = _queue, options) do
-    default_type = if match?([_ | _], options[:headers]), do: :headers, else: :direct
+    default_type = if headers?(options), do: :headers, else: :direct
     exchange_type = options[:exchange_type] || default_type
     exchange_name = options[:exchange] || "amq.#{exchange_type}"
     {exchange_name, exchange_type}
@@ -169,14 +169,23 @@ defmodule MessageQueue.Adapters.RabbitMQ.ProducerWorker do
     {exchange_name, exchange_type}
   end
 
-  defp get_routing_key("amq.headers", _queue, _options), do: ""
+  defp get_routing_key("amq.headers", _queue, options) do
+    if keep_routing_key?(options), do: Keyword.get(options, :routing_key, ""), else: ""
+  end
 
   defp get_routing_key(_exchange, queues, options) when is_list(queues) do
     Keyword.get(options, :routing_key, "")
   end
 
+  # Headers exchanges route on headers, so the routing key is normally cleared.
+  # Pass `keep_routing_key: true` to preserve it (e.g. to still reach a queue
+  # bound by routing key while also carrying headers).
   defp get_routing_key(_exchange, queue, options) do
-    if match?([_ | _], options[:headers]), do: "", else: Keyword.get(options, :routing_key, queue)
+    if headers?(options) and not keep_routing_key?(options) do
+      ""
+    else
+      Keyword.get(options, :routing_key, queue)
+    end
   end
 
   defp declare_and_publish(channel, message, options) do
@@ -229,6 +238,10 @@ defmodule MessageQueue.Adapters.RabbitMQ.ProducerWorker do
     queue_module().bind(channel, queue, exchange, options)
   end
 
+  defp headers?(options), do: match?([_ | _], options[:headers])
+
+  defp keep_routing_key?(options), do: options[:keep_routing_key] == true
+
   defp encode_message(message, opts) do
     Message.encode(message,
       type: opts[:message_type],

test/message_queue/adapters/rabbitmq/producer_worker_test.exs

@@ -325,6 +325,66 @@ defmodule MessageQueue.Adapters.RabbitMQ.ProducerWorkerTest do
     assert <<131, _rest::binary>> = bin_payload
   end
 
+  test "request publish with headers clears the routing key by default" do
+    assert :ok =
+             ProducerWorker.request(:test_channel, {
+               :publish,
+               %{id: 42},
+               "known.queue",
+               [message_type: :json, headers: [{"x-match", "all"}]]
+             })
+
+    assert_received {:publish, :test_channel, exchange, "", _payload, _options}
+    refute exchange == ""
+  end
+
+  test "request publish with headers and keep_routing_key: true preserves the routing key" do
+    assert :ok =
+             ProducerWorker.request(:test_channel, {
+               :publish,
+               %{id: 42},
+               "known.queue",
+               [message_type: :json, headers: [{"x-match", "all"}], keep_routing_key: true]
+             })
+
+    assert_received {:publish, :test_channel, "", "known.queue", _payload, _options}
+  end
+
+  test "keep_routing_key: true preserves an explicit routing key on the headers exchange" do
+    assert :ok =
+             ProducerWorker.request(:test_channel, {
+               :publish,
+               %{id: 42},
+               "",
+               [
+                 message_type: :json,
+                 exchange: "amq.headers",
+                 headers: [{"x-match", "all"}],
+                 routing_key: "keep.me",
+                 keep_routing_key: true
+               ]
+             })
+
+    assert_received {:publish, :test_channel, "amq.headers", "keep.me", _payload, _options}
+  end
+
+  test "headers exchange clears the routing key without keep_routing_key" do
+    assert :ok =
+             ProducerWorker.request(:test_channel, {
+               :publish,
+               %{id: 42},
+               "",
+               [
+                 message_type: :json,
+                 exchange: "amq.headers",
+                 headers: [{"x-match", "all"}],
+                 routing_key: "ignored"
+               ]
+             })
+
+    assert_received {:publish, :test_channel, "amq.headers", "", _payload, _options}
+  end
+
   defp restore_env(key, nil), do: Application.delete_env(:message_queue, key)
   defp restore_env(key, value), do: Application.put_env(:message_queue, key, value)
 end