[rabbitmq-discuss] Bug with consistent hash exchange.

Thu Dec 8 15:02:10 GMT 2011

On Thu, Dec 08, 2011 at 06:34:24AM -0800, Wojciech Durczyński wrote:
> Let's consider following routing:
> 
> exchange A of type topic
> exchange B of type consistent-hash-exchange
> exchange C of type consistent-hash-exchange
> 
> binding A -> B with routing key "b.#"
> binding A -> C with routing key "c.#"
> 
> If there are queues bound to exchanges B and C, everything works as 
> expected.
> 
> If there aren't queues bound to C, but there are queues bound to B then 
> message with routing key "c.x" sent to exchange A is routed to one of the 
> B's queues. It should be discarded instead or cause NO-ROUTE or 
> NO_CONSUMERS (message flag dependent).

Agreed. I've just knocked up the following test, and you're right. It's
clearly wrong. Sorry about that. I'll get it fixed and it should be in
the next release.

-module(test).
-compile([export_all]).

-include_lib("amqp_client/include/amqp_client.hrl").

test_consistent_hash() ->
    A = <<"A">>, B = <<"B">>, C = <<"C">>, Q = <<"my_queue">>,
    {ok, Conn} = amqp_connection:start(#amqp_params_network{}),
    {ok, Chan} = amqp_connection:open_channel(Conn),
    #'queue.declare_ok' {} = amqp_channel:call(Chan, #'queue.declare'{ queue = Q }),
    #'exchange.declare_ok'{} =
        amqp_channel:call(
          Chan, #'exchange.declare'{ exchange = A, type = <<"topic">> }),
    #'exchange.declare_ok'{} =
        amqp_channel:call(
          Chan, #'exchange.declare'{ exchange = B, type = <<"x-consistent-hash">> }),
    #'exchange.declare_ok'{} =
        amqp_channel:call(
          Chan, #'exchange.declare'{ exchange = C, type = <<"x-consistent-hash">> }),
    #'exchange.bind_ok'{} = amqp_channel:call(
                              Chan, #'exchange.bind'{ source = A, destination = B,
                                                      routing_key = <<"b.#">> }),
    #'exchange.bind_ok'{} = amqp_channel:call(
                              Chan, #'exchange.bind'{ source = A, destination = C,
                                                      routing_key = <<"c.#">> }),
    #'queue.bind_ok'{} = amqp_channel:call(
                           Chan, #'queue.bind'{ queue = Q, exchange = B,
                                                routing_key = <<"10">> }),
    Msg = #amqp_msg { props = #'P_basic'{}, payload = <<"hello">> },
    Publish = #'basic.publish'{ routing_key = <<"c.foo">>, exchange = A },
    amqp_channel:call(Chan, Publish, Msg),
    timer:sleep(1000),
    io:format("~p~n", [amqp_channel:call(Chan, #'queue.declare'{ queue = Q })]),
    amqp_connection:close(Conn).

Matthew