A question about the headers exchange routing

[everttimmer1963]

Describe the bug

Problem

The header exchange does not route messages as advertised. Since criteria for routing are associated with a queue-binding, I expect only messages to be queued to this queue when the message headers match the criteria stored with the queue binding. This, however, is not the reality.

When configuring an exchange with 2 bound queues, each with their own criteria, and publishing 2 messages to the exchange, each with an argument that matches one of the binding criteria, I would expect that the message that has arguments matching the binding criteria for queue one, is only routed to queue one, and that the message that matches the criteria of the binding for queue two, would be routed to queue two only.

Both messages, however, are routed to both queues, comletely ignoring the criteria.

Below a unit test (C#) that produces this error.

`
[Fact]
public async Task When_Declaring_Headers_Exchange_And_Publishing_To_Exchange_Then_Only_Queues_With_Matching_Headers_Should_Receive_Message()
{
// Arrange
var exchangeName = "test-xchg";
var queueName1 = "queue-one";
var queueName2 = "queue-two";
var connection = await factory.CreateConnectionAsync();
var channel = await connection.CreateChannelAsync();
var properties1 = new BasicProperties
{
ContentType = "text/plain",
DeliveryMode = DeliveryModes.Persistent, // Persistent
Priority = 0,
Headers = new Dictionary<string, object?>
{
{ "x-custom-header", "custom-value-1" }
}
};
var arguments1 = new Dictionary<string, object?> { { "x-match", "all" }, { "x-custom-header", "custom-value-1" } };
var properties2 = new BasicProperties
{
ContentType = "text/plain",
DeliveryMode = DeliveryModes.Persistent, // Persistent
Priority = 0,
Headers = new Dictionary<string, object?>
{
{ "x-custom-header", "custom-value-2" }
}
};
var arguments2 = new Dictionary<string, object?> { { "x-match", "all" }, { "x-custom-header", "custom-value-2" } };

    // Act
    await channel.ExchangeDeclareAsync(exchangeName, ExchangeType.Headers, durable: true, autoDelete: false, arguments: null, noWait: false, cancellationToken: default);
    await channel.QueueDeclareAsync(queueName1, durable: true, exclusive: false, autoDelete: false, arguments: null, noWait: false, cancellationToken: default);
    await channel.QueueBindAsync(queueName1, exchangeName, routingKey: "", arguments: arguments1, noWait: false, cancellationToken: default);
    await channel.QueueDeclareAsync(queueName2, durable: true, exclusive: false, autoDelete: false, arguments: null, noWait: false, cancellationToken: default);
    await channel.QueueBindAsync(queueName2, exchangeName, routingKey: "", arguments: arguments2, noWait: false, cancellationToken: default);
    var message1 = "Hello World! custom-value-1";
    var message2 = "Hello World! custom-value-2";
    var body1 = Encoding.UTF8.GetBytes(message1);
    var body2 = Encoding.UTF8.GetBytes(message2);

    await channel.BasicPublishAsync(exchangeName, routingKey: "", false, basicProperties: properties1, body: body1);
    await channel.BasicPublishAsync(exchangeName, routingKey: "", false, basicProperties: properties2, body: body2);

    var receivedOne = false;
    var receivedTwo = false;
    while (true)
    {
        var result = await channel.BasicGetAsync(queueName1, true);
        if (result is not null)
        {
            receivedOne = Encoding.UTF8.GetString(result.Body.ToArray()) == "Hello World! custom-value-1";
        }
        result = await channel.BasicGetAsync(queueName2, true);
        if (result is not null)
        {
            receivedTwo = Encoding.UTF8.GetString(result.Body.ToArray()) == "Hello World! custom-value-2";
        }
        if (receivedOne && receivedTwo)
        {
            break;
        }
        await Task.Delay(100);
    }

    var count1 = await channel.MessageCountAsync(queueName1, cancellationToken: default);
    var count2 = await channel.MessageCountAsync(queueName2, cancellationToken: default);

    // Assert
    Assert.True(count1 + count2 == 0);

    // Clean-up
    await channel.QueueDeleteAsync(queueName1, ifUnused: false, ifEmpty: false, cancellationToken: default);
    await channel.QueueDeleteAsync(queueName2, ifUnused: false, ifEmpty: false, cancellationToken: default);
    await channel.ExchangeDeleteAsync(exchangeName, ifUnused: false, cancellationToken: default);
    await channel.CloseAsync();
    await channel.DisposeAsync();
    await connection.CloseAsync();
    await connection.DisposeAsync();
}

`

Reproduction steps

1. Declare exchange with name "test-xchg" of type "headers"
2. Declare a Queue named "Queue-One" and bind to exchange with arguments: "x-match=all", "x-header=header-one"
3. Declare a Queue named "Queue-Two" and bind to exchange with arguments "x-match=all", "s-header=header-two"
4. Publish a message to the exchange "test-xchg" with argument "x-header=header-one"
5. Publish a message to the exchange "test-xchg" with argument "x-header=header-two"
   ...

Expected behavior

I expect message 1 only to appear in queue-one
I expect message 2 only to appear in queue-two

Additional context

No response

[michaelklishin]

@everttimmer1963 as advertised where?

I cannot reproduce against RabbitMQ 4.1.0 using the following example from one of the client library doc guides.

Documentation Example

#!/usr/bin/env ruby
# encoding: utf-8

require "rubygems"
require "bunny"

puts "=> Headers exchange routing"
puts

conn = Bunny.new
conn.start

ch   = conn.create_channel
x    = ch.headers("headers")

q1   = ch.queue("rabbitmq-server.13934.q1.linux", exclusive: true).bind(x, arguments: {"os" => "linux", "cores" => 8, "x-match" => "all"})
q2   = ch.queue("rabbitmq-server.13934.q2.macos", exclusive: true).bind(x, arguments: {"os" => "macos",   "cores" => 4, "x-match" => "any"})

q1.subscribe do |delivery_info, properties, content|
  puts "#{q1.name} received #{content}"
end
q2.subscribe do |delivery_info, properties, content|
  puts "#{q2.name} received #{content}"
end

x.publish("8 cores/Linux", :headers => {"os" => "linux", "cores" => 8})
x.publish("8 cores/OS X",  :headers => {"os" => "macos",   "cores" => 8})
x.publish("4 cores/Linux", :headers => {"os" => "linux", "cores" => 4})

sleep 0.5
conn.close

Which outputs

=> Headers exchange routing

rabbitmq-server.13934.q1.linux received 8 cores/Linux
rabbitmq-server.13934.q2.macos received 8 cores/OS X
rabbitmq-server.13934.q2.macos received 4 cores/Linux

As is very common to see with the (rare) headers exchange questions, it likely comes down to

1. What x-match value is used. The exchange, sadly, supports four different values: all, any, all-with-x and any-with-x
2. The structure of the headers

Existing Tests That Use the Headers Exchange

RabbitMQ has tests for the headers exchange behavior, such as

%% Test that headers exchange's x-match binding argument set to all-with-x and any-with-x
%% works as expected. The use case being tested here:
%% Route dead-letter messages to different target queues
%% according to first death reason and first death queue.
dead_letter_headers_first_death_route(Config) ->
    {_Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
    QName1 = ?config(queue_name, Config),
    QName2 = <<"dead_letter_headers_first_death_route_source_queue_2">>,
    DLXExpiredQName = ?config(queue_name_dlx, Config),
    DLXRejectedQName = ?config(queue_name_dlx_2, Config),
    Args = ?config(queue_args, Config),
    Durable = ?config(queue_durable, Config),
    DLXExchange = ?config(dlx_exchange, Config),

    #'exchange.declare_ok'{} = amqp_channel:call(Ch, #'exchange.declare'{exchange = DLXExchange,
                                                                         type = <<"headers">>}),
    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName1,
                                                                   arguments = [{<<"x-dead-letter-exchange">>, longstr, DLXExchange} | Args],
                                                                   durable = Durable}),
    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = QName2,
                                                                   arguments = [{<<"x-dead-letter-exchange">>, longstr, DLXExchange} | Args],
                                                                   durable = Durable}),
    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = DLXExpiredQName,
                                                                   durable = Durable}),
    #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = DLXRejectedQName,
                                                                   durable = Durable}),
    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue       = DLXExpiredQName,
                                                             exchange    = DLXExchange,
                                                             arguments = [{<<"x-match">>, longstr, <<"all-with-x">>},
                                                                          {<<"x-first-death-reason">>, longstr, <<"expired">>},
                                                                          {<<"x-first-death-queue">>, longstr, QName1}]
                                                            }),
    #'queue.bind_ok'{} = amqp_channel:call(Ch, #'queue.bind'{queue       = DLXRejectedQName,
                                                             exchange    = DLXExchange,
                                                             arguments = [{<<"x-match">>, longstr, <<"any-with-x">>},
                                                                          {<<"x-first-death-reason">>, longstr, <<"rejected">>}]
                                                            }),
    %% Send 1st message to 1st source queue and let it expire.
    P1 = <<"msg1">>,
    amqp_channel:call(Ch, #'basic.publish'{routing_key = QName1},
                      #amqp_msg{payload = P1,
                                props = #'P_basic'{expiration = <<"0">>}}),
    %% The 1st message gets dead-lettered to DLXExpiredQName.
    wait_for_messages(Config, [[DLXExpiredQName, <<"1">>, <<"1">>, <<"0">>]]),
    _ = consume(Ch, DLXExpiredQName, [P1]),
    consume_empty(Ch, DLXExpiredQName),
    wait_for_messages(Config, [[QName1, <<"0">>, <<"0">>, <<"0">>]]),
    %% Send 2nd message to 2nd source queue and let it expire.
    P2 = <<"msg2">>,
    amqp_channel:call(Ch, #'basic.publish'{routing_key = QName2},
                      #amqp_msg{payload = P2,
                                props = #'P_basic'{expiration = <<"0">>}}),
    %% Send 2nd message should not be routed by the dead letter headers exchange.
    rabbit_ct_helpers:consistently(?_assertEqual(#'basic.get_empty'{},
                                                 amqp_channel:call(Ch, #'basic.get'{queue = DLXExpiredQName}))),
    %% Send and reject the 3rd message.
    P3 = <<"msg3">>,
    publish(Ch, QName2, [P3]),
    case group_name(Config) of
        at_most_once ->
            wait_for_messages(Config, [[QName2, <<"1">>, <<"1">>, <<"0">>]]);
        at_least_once ->
            wait_for_messages(Config, [[QName2, <<"2">>, <<"1">>, <<"0">>]])
    end,
    [DTag] = consume(Ch, QName2, [P3]),
    amqp_channel:cast(Ch, #'basic.reject'{delivery_tag = DTag,
                                          requeue      = false}),
    %% The 3rd message gets dead-lettered to DLXRejectedQName.
    wait_for_messages(Config, [[DLXRejectedQName, <<"1">>, <<"1">>, <<"0">>]]),
    _ = consume(Ch, DLXRejectedQName, [P3]),
    consume_empty(Ch, DLXRejectedQName),
    _ = amqp_channel:call(Ch, #'queue.delete'{queue = QName2}),
    ok.

Or this:

    @Test
    @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_10)
    public void headersWithXRouting() throws Exception {
        Map<String, Object> spec = new HashMap<String, Object>();
        spec.put("x-key-1", "value-1");
        spec.put("x-key-2", "value-2");
        spec.put("x-match", "all-with-x");
        channel.queueBind(Q1, "amq.match", "", spec);
        spec.put("x-match", "any-with-x");
        channel.queueBind(Q2, "amq.match", "", spec);

        AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder();
        channel.basicPublish("amq.match", "", props.build(), "0".getBytes());

        Map<String, Object> map = new HashMap<String, Object>();
        props.headers(map);

        map.clear();
        map.put("x-key-1", "value-1");
        channel.basicPublish("amq.match", "", props.build(), "1".getBytes());

        map.clear();
        map.put("x-key-1", "value-1");
        map.put("x-key-2", "value-2");
        channel.basicPublish("amq.match", "", props.build(), "2".getBytes());

        map.clear();
        map.put("x-key-1", "value-1");
        map.put("x-key-2", "value-2");
        map.put("x-key-3", "value-3");
        channel.basicPublish("amq.match", "", props.build(), "3".getBytes());

        checkGet(Q1, true); // 2
        checkGet(Q1, true); // 3
        checkGet(Q1, false);

        checkGet(Q2, true); // 1
        checkGet(Q2, true); // 2
        checkGet(Q2, true); // 3
        checkGet(Q2, false);
    }
    
    private void checkGet(String queue, boolean messageExpected)
        throws IOException
    {
        GetResponse r = channel.basicGet(queue, true);
        if (messageExpected) {
            assertNotNull(r);
        } else {
            assertNull(r);
        }
    }    

[everttimmer1963]

Thanks for your work,

This was tested against the most recent .NET RabbitMQ.Client

I have written a Mock library for it because the existing mock libraries are outdated.

See: https://www.nuget.org/packages/RabbitMQ.Client.Mock

[michaelklishin]

@everttimmer1963 thank you for putting in the time! Please announce it on our Discord server, in the discussions (of the .NET client, I guess) and on the mailing list :)

[michaelklishin]

Finally, headers exchanges are very rarely used because they usually can be swapped for topic exchanges using a basic convention.

"a = 1 AND b = 2" (two header conditions) can be expressed as a topic, a.1.b.2 or even a.1.and.b.2, and the topic exchanges have infinitely more mileage on them.

We don't have neither RabbitMQ nor .NET client details, so I have provided two tests and one doc example that demonstrate the "selective routing" you are looking for.

[everttimmer1963]

Thanks for replying.

I know that headers exchanges are rarely being used. I have written a RabbitMQ mock library and I want it to cover as much scenarios as possible.

[ansd]

You need to use value all-with-x as explained in https://www.rabbitmq.com/tutorials/amqp-concepts#exchange-headers :

For "any" and "all", headers beginning with the string x- will not be used to evaluate matches. Setting "x-match" to "any-with-x" or "all-with-x" will also use headers beginning with the string x- to evaluate matches.

[everttimmer1963]

Thanks! That was the solution. I read about the expected behavior but missed that remark.

I have written a RabbitMQ.Client.Mock library for unit-tests usage that mocks behavior (at least tries to) of RabbitMQ and could not get this one to do what i expected it to do.

See https://www.nuget.org/packages/RabbitMQ.Client.Mock for information.