RabbitMQ for Windows: Direct Exchanges

Posts In This Series

This is the fifth installment to the series: RabbitMQ for Windows.  In the last installment, we took a look at the four exchange types provided by RabbitMQ: Direct, Fanout, Topic, and Headers.  In this installment we’ll walk through an example which uses a direct exchange type directly and we’ll take a look at the push API.

In the Hello World example from the second installment of the series, we used a direct exchange type implicitly by taking advantage of the automatic  binding of queues to the default exchange using the queue name as the routing key.  The example we’ll work through this time will be similar, but we’ll declare and bind to the exchange explicitly.

This time our example will be a distributed logging application.  We’ll create a Producer console application which publishes a logging message for some noteworthy action and a Consumer console application which displays the message to the console.

Beginning with our Producer app, we’ll start by establishing a connection using the default settings, create the connection, and create a channel:

using RabbitMQ.Client;

namespace Producer
{
  class Program
  {
    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();
    }
  }
}

Next, we need to declare the exchange we’ll be publishing our message to.  We need to give our exchange a name in order to reference it later, so let’s use “direct-exchange-example”:

channel.ExchangeDeclare("direct-exchange-example", ExchangeType.Direct);

The second parameter indicates the exchange type.  For the official RabbitMQ .Net client, this is just a simple string containing one of the values: direct, fanout, topic, or headers.  The type RabbitMQ.Client.ExchangeType defines each of the exchange types as a constant for convenience.

Next, let’s call some method which might produce a value worthy of interest.  We’ll call the method DoSomethingInteresting() and have it return a string value:

string value = DoSomethingInteresting();

For the return value, the implementation of DoSomethingInteresting() can just return the string value of a new Guid:

static string DoSomethingInteresting() 
{ 
  return Guid.NewGuid().ToString(); 
} 

Next, let’s use the returned value to create a log message containing a severity level of Information:

string logMessage = string.Format("{0}: {1}", TraceEventType.Information, value); 

Next, we need to convert our log message to a byte array and publish the message to our new exchange:

  byte[] message = Encoding.UTF8.GetBytes(logMessage);
  channel.BasicPublish("direct-exchange-example", "", null, message); 
  

Here, we use an empty string as our routing key and null for our message properties.

We end our Producer by closing the channel and connection:

channel.Close(); 
connection.Close(); 

Here’s the full listing:

using System;
using System.Diagnostics;
using System.Text;
using System.Threading;
using RabbitMQ.Client;

namespace Producer
{
  class Program
  {
    static void Main(string[] args)
    {
      Thread.Sleep(1000);
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();

      channel.ExchangeDeclare("direct-exchange-example", ExchangeType.Direct);
      string value = DoSomethingInteresting();
      string logMessage = string.Format("{0}: {1}", TraceEventType.Information, value);

      byte[] message = Encoding.UTF8.GetBytes(logMessage);
      channel.BasicPublish("direct-exchange-example", "", null, message);

      channel.Close();
      connection.Close();
    }

    static string DoSomethingInteresting()
    {
      return Guid.NewGuid().ToString();
    }
  }
}

Note that our logging example’s Producer differs from our Hello World’s Producer in that we didn’t declare a queue this time.  In our Hello World example, we needed to run our Producer before the Consumer since the Consumer simply retrieved a single message and exited.  Had we published to the default exchange without declaring the queue first, our message would simply have been discarded by the server before the Consumer had an opportunity to declare and bind the queue.

Next, we’ll create our Consumer which starts the same way as our Producer code:

using RabbitMQ.Client;

namespace Consumer
{
  class Program
  {
    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();

      channel.ExchangeDeclare("direct-exchange-example", ExchangeType.Direct);
    }
  }
}

Next, we need to declare a queue to bind to our exchange.  Let’s name our queue “logs”:

channel.QueueDeclare("logs", false, false, true, null); 

To associate our logs queue with our exchange, we use the QueueBind() method providing the name of the queue, the name of the exchange, and the binding key to filter messages on:

channel.QueueBind("logs", "direct-exchange-example", "");

At this point we could consume messages using the pull API method BasicGet() as we did in the Hello World example, but this time we’ll use the push API.  To have messages pushed to us rather than us pulling messages, we first need to declare a consumer:

var consumer = new QueueingBasicConsumer(channel);

To start pushing messages to our consumer, we call the channel’s BasicConsume() method and tell it which consumer to start pushing messages to:

channel.BasicConsume(“logs”, true, consumer); 

Here, we specify the queue to consume messages from, a boolean flag instructing messages to be auto-acknowledged (see discussion in the Getting the Message section of Hello World Review), and the consumer to push the messages to.  

Now, any messages placed on the queue will automatically be retrieved and placed in a local in-memory queue.  To dequeue a message from the local queue, we call the Dequeue() method on the consumer’s Queue property:

var eventArgs = (BasicDeliverEventArgs)consumer.Queue.Dequeue(); 

This method call blocks until a message is available to be dequeued, or until an EndOfStreamException is thrown indicating that the consumer was cancelled, the channel was closed, or the connection otherwise was terminated.

Once the Dequeue() method returns, the BasicDeliverEventArgs contains the bytes published from the Producer in the Body property, so we can convert this value back into a string and print it to the console:

var message = Encoding.UTF8.GetString(eventArgs.Body);
Console.WriteLine(message); 

We end our Consumer by closing the channel and connection:

channel.Close();
connection.Close(); 

Here’s the full listing:

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Consumer
{
  class Program
  {
    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();

      channel.ExchangeDeclare("direct-exchange-example", ExchangeType.Direct);
      channel.QueueDeclare("logs", false, false, true, null);
      channel.QueueBind("logs", "direct-exchange-example", "");

      var consumer = new QueueingBasicConsumer(channel);
      channel.BasicConsume("logs", true, consumer);

      var eventArgs = (BasicDeliverEventArgs) consumer.Queue.Dequeue();

      string message = Encoding.UTF8.GetString(eventArgs.Body);
      Console.WriteLine(message);

      channel.Close();
      connection.Close();
      Console.ReadLine();
    }
  }
}

If we run the resulting Consumer.exe at this point, it will block until a message is routed to the queue.  Running the Producer.exe from another shell produces a message on the consumer console similar to the following:

Information: 610fe447-bf31-41d2-ae29-414b2d00087b 

That concludes our direct exchange example.  Next time, we’ll take a look at the Fanout exchange type.