RabbitMQ Playground!

Headers exchange play in order to understand them.

RabbitPlay.Helper.test_publishes/0

This function implements an example similar to the headers exchange explained in this post

How Headers Exchange works

This is a simple exercise to learn how to use headers exchange.

First of all, declaration:

Exchange.declare(chan, exchange_name, type, options)

We’ll need a channel chan, an exchange name exchange_name, and a type of exchange type, options are optional, for example:

Exchange.declare(chan, "test_headers", :headers, durable: true)

In order to obtain the channel we need a connection, all this is obtained with:

use AMQP

{:ok, conn} = Connection.open("amqp://guest:guest@localhost")
{:ok, chan} = Channel.open(conn)

(Supposing that we have a RabbitMQ server in localhost with the user guest and password guest)

After that, we need some queues, lets say:

Queue.declare(chan, "queue_a", durable: true)
Queue.declare(chan, "queue_b", durable: true)

Now for the bindings, and here is where the interesting part starts! In the binding we define which type of message (according to the headers) we want to pass to the queue, lets see an example of a binding:

The binding function looks like this:

Queue.bind(chan, queue_name, exchange_name, options)

This binds the queue named queue_name to the exchange exchange_name with the options options, here is a concrete example:

Queue.bind(
  chan,
  "queue_a",
  "test_headers",
  arguments: [{"format", "pdf"}, {"type", "log"}, {"x-match", "any"}]
)

Now we have our queue_a binded to the exchange test_headers with these arguments:

[{"format", "pdf"}, {"type", "log"}, {"x-match", "any"}]

The x-match parameter can only be any or all:

• any: lets in any message that contains any of the parameters described. For example: A message with headers -> [{"format", "pdf"}], will pass. But a message with headers -> [{"format", "invent"}, {"param", "random"}] will not, because “format” is not “pdf” and “param” is not even listed on the binding.
• all: forces that the message has all the parameters that the binding has. For example: A message with headers -> [{"format", "pdf"}, {"type", "log"}] will pass, because “x-match” is always ignored by amqp. A message with headers -> [{"format", "zip"}] will not pass, because neither “format” is “pdf” or “type” is specified.

Lets bind “queue_b” and send some examples:

Queue.bind(
  chan,
  "queue_b",
  "test_headers",
  arguments: [{"format", "zip"}, {"type", "log"}, {"x-match", "all"}]
)

These are the exchange bindings:

So, if we send a message with [{"format", "pdf"}] it will only go to “queue_a” and if we send a message with [{"format", "zip"}, {"type", "log"}] it will be sent to both queues, lets see this in action.

First send to queue_a only:

The publish function looks like this:

Basic.publish(chan, exchange_name, routing_key, message_text, options)

Here is a concrete example:

Basic.publish(chan, "test_headers", "", "Hello there", headers: [{"format", "pdf"}])

(Note that routing_key is “”, since we are using a headers exchange amqp will ignore the routing key)

This only arrives to queue_a:

(Trust me is queue_a)

Now lets try sending the other message:

Basic.publish(
  chan,
  "test_headers",
  "",
  "Hello there second time",
  headers: [{"format", "zip"}, {"type", "log"}]
)

This will be queue_a:

And this will be queue_b:

NICE! Now experiment yourself with the headers exchange and let me now if this helped you :D