Headers exchanges examine the message headers to determine which queues a message should be routed to.  As discussed earlier in this series, headers exchanges are similar to topic exchanges in that they allow you to specify multiple criteria, but offer a bit more flexibility in that the headers can be constructed using a wider range of data types (1).

To subscribe to receive messages from a headers exchange, a dictionary of headers is specified as part of the binding arguments.  In addition to the headers, a key of “x-match” is also included in the dictionary with a value of “all”, specifying that messages must be published with all the specified headers in order to match, or “any”, specifying that the message needs to only have one of the specified headers specified.

As our final example, we’ll create a Producer application which publishes the message “Hello, World!” using a headers exchange.  Here’s our Producer code:

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Framing.v0_9_1;

namespace Producer
{
  class Program
  {
    const string ExchangeName = "header-exchange-example";

    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      connectionFactory.HostName = "localhost";

      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();
      channel.ExchangeDeclare(ExchangeName, ExchangeType.Headers, false, true, null);
      byte[] message = Encoding.UTF8.GetBytes("Hello, World!");

      var properties = new BasicProperties();
      properties.Headers = new Dictionary<string, object>();
      properties.Headers.Add("key1", "12345");
      
      TimeSpan time = TimeSpan.FromSeconds(10);
      var stopwatch = new Stopwatch();
      Console.WriteLine("Running for {0} seconds", time.ToString("ss"));
      stopwatch.Start();
      var messageCount = 0;

      while (stopwatch.Elapsed < time)
      {
        channel.BasicPublish(ExchangeName, "", properties, message);
        messageCount++;
        Console.Write("Time to complete: {0} seconds - Messages published: {1}\r", (time - stopwatch.Elapsed).ToString("ss"), messageCount);
        Thread.Sleep(1000);
      }

      Console.Write(new string(' ', 70) + "\r");
      Console.WriteLine("Press any key to exit");
      Console.ReadKey();
      message = Encoding.UTF8.GetBytes("quit");
      channel.BasicPublish(ExchangeName, "", properties, message);
      connection.Close();
    }
  }
}

In the Producer, we’ve used a generic dictionary of type Dictionary<string, object> and added a single key “key1” with a value of “12345”.  As with our previous example, we’re using a stopwatch as a way to publish messages continually for 10 seconds.

For our Consumer application, we can use an “x-match” argument of “all” with the single key/value pair specified by the Producer, or we can use an “x-match” argument of “any” which includes the key/value pair specified by the Producer along with other potential matches.  We’ll use the latter for our example.   Here’s our Consumer code:

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

namespace Consumer
{
  class Program
  {
    const string QueueName = "header-exchange-example";
    const string ExchangeName = "header-exchange-example";

    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      connectionFactory.HostName = "localhost";

      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();
      channel.ExchangeDeclare(ExchangeName, ExchangeType.Headers, false, true, null);
      channel.QueueDeclare(QueueName, false, false, true, null);

      IDictionary specs = new Dictionary();
      specs.Add("x-match", "any");
      specs.Add("key1", "12345");
      specs.Add("key2", "123455");
      channel.QueueBind(QueueName, ExchangeName, string.Empty, specs);

      channel.StartConsume(QueueName, MessageHandler);
      connection.Close();
    }

    public static void MessageHandler(IModel channel, DefaultBasicConsumer consumer, BasicDeliverEventArgs eventArgs)
    {
      string message = Encoding.UTF8.GetString(eventArgs.Body);
      Console.WriteLine("Message received: " + message);
      foreach (object headerKey in eventArgs.BasicProperties.Headers.Keys)
      {
        Console.WriteLine(headerKey + ": " + eventArgs.BasicProperties.Headers[headerKey]);
      }

      if (message == "quit")
        channel.BasicCancel(consumer.ConsumerTag);
    }
  }
}

Rather than handling our messages inline as we’ve done in previous examples, this example uses an extension method named StartConsume() which accepts a callback to be invoked each time a message is received.  Here’s the extension method used by our example:

using System;
using System.IO;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Consumer
{
  public static class ChannelExtensions
  {
    public static void StartConsume(this IModel channel, string queueName,  Action<IModel, DefaultBasicConsumer, BasicDeliverEventArgs> callback)
    {
      var consumer = new QueueingBasicConsumer(channel);
      channel.BasicConsume(queueName, true, consumer);

      while (true)
      {
        try
        {
          var eventArgs = (BasicDeliverEventArgs)consumer.Queue.Dequeue();
          new Thread(() => callback(channel, consumer, eventArgs)).Start();
        }
        catch (EndOfStreamException)
        {
          // The consumer was cancelled, the model closed, or the connection went away.
          break;
        }
      }
    }
  }
}

Setting our solution to run both the Producer and Consumer applications upon startup, running our example produces output similar to the following:

Producer

Running for 10 seconds
Time to complete: 08 seconds - Messages published: 2

Consumer

Message received: Hello, World!
key1: 12345
Message received: Hello, World!
key1: 12345

That concludes our headers exchange example as well as the RabbitMQ for Windows series.  For more information on working with RabbitMQ, see the documentation at http://www.rabbitmq.com or the purchase the book RabbitMQ in Action by Alvaro Videla and Jason Williams.  I hope you enjoyed the series.

Footnotes:

1 – See http://www.rabbitmq.com/amqp-0-9-1-errata.html#section_3 and http://hg.rabbitmq.com/rabbitmq-dotnet-client/diff/4def852523e2/projects/client/RabbitMQ.Client/src/client/impl/WireFormatting.cs for supported field types.

Topic exchanges are similar to direct exchanges in that they use a routing key to determine which queue a message should be delivered to, but they differ in that they provide the ability to match on portions of a routing key.  When publishing to a topic exchange, a routing key consisting of multiple words separated by periods (e.g. “word1.word2.word3”) will be matched against a pattern supplied by the binding queue.  Patterns may contain an asterisk (“*”) to match a word in a specific segment or a hash (“#”) to match zero or more words.  As discussed earlier in the series, the topic exchange type can be useful for directing messages based on multiple categories or for routing messages originating from multiple sources.

To demonstrate topic exchanges, we’ll return to our logging example, but this time we’ll subscribe to a subset of the messages being published to demonstrate the flexibility of how routing keys are used by topic exchanges.  For this example, we’ll be modeling a scenario where a company may have multiple client installations, each of which may be used to service different sectors of a company’s business model (e.g. Business or Personal sectors).  We’ll use a routing key that specifies the sector and subscribe to messages published for the Personal sector only.

As with our previous examples, we’ll keep things simple by creating console applications for a Producer and a Consumer.  Let’s start by creating the Producer app and establishing a connection using the default settings:

using RabbitMQ.Client;

namespace Producer
{
  class Program
  {
    const long ClientId = 10843;

    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
    }
  }
}

Rather than just publishing messages directly from the Main() method as with our first logging example, let’s create a separate logger object this time.  Here the logger interface and implementation we’ll be using:

  interface ILogger
  {
    void Write(Sector sector, string entry, TraceEventType traceEventType);
  }

  class RabbitLogger : ILogger, IDisposable
  {
    readonly long _clientId;
    readonly IModel _channel;
    bool _disposed;

    public RabbitLogger(IConnection connection, long clientId)
    {
      _clientId = clientId;
      _channel = connection.CreateModel();
      _channel.ExchangeDeclare("direct-exchange-example", ExchangeType.Topic, false, true, null);
    }

    public void Dispose()
    {
      if (!_disposed)
      {
        if (_channel != null && _channel.IsOpen)
        {
          _channel.Close();
        }
      }
      GC.SuppressFinalize(this);
    }

    public void Write(Sector sector, string entry, TraceEventType traceEventType)
    {
      byte[] message = Encoding.UTF8.GetBytes(entry);
      string routingKey = string.Format("{0}.{1}.{2}", _clientId, sector.ToString(), traceEventType.ToString());
      _channel.BasicPublish("topic-exchange-example", routingKey, null, message);
    }

    ~RabbitLogger()
    {
      Dispose();
    }
  }

In addition to an open IConnection, our RabbitLogger class is instantiated with a client Id.  We use this as part of the routing key.  Since each log can vary by sector, we pass a Sector enum as part of the Write() method.  Here’s our Sector enum:

  public enum Sector
  {
    Personal,
    Business
  }

Returning to our Main() method, we now need to instantiate our RabbitLogger and log messages with differing sectors.  As as way to ensure our client has an opportunity to subscribe to our messages and to help emulate a continual stream of log messages being published, let’s use the logger to publish a series of log messages every second for 10 seconds:

      TimeSpan time = TimeSpan.FromSeconds(10);
      var stopwatch = new Stopwatch();
      Console.WriteLine("Running for {0} seconds", time.ToString("ss"));
      stopwatch.Start();

      while (stopwatch.Elapsed < time)
      {
        using (var logger = new RabbitLogger(connection, ClientId))
        {
          Console.Write("Time to complete: {0} seconds\r", (time - stopwatch.Elapsed).ToString("ss"));
          logger.Write(Sector.Personal, "This is an information message", TraceEventType.Information);
          logger.Write(Sector.Business, "This is an warning message", TraceEventType.Warning);
          logger.Write(Sector.Business, "This is an error message", TraceEventType.Error);
          Thread.Sleep(1000);
        }
      }

This code prints out the time remaining just to give us a little feedback on the publishing progress.  Finally, we’ll close our our connection and prompt the user to exit the console application:

      connection.Close();
      Console.Write("                             \r");
      Console.WriteLine("Press any key to exit");
      Console.ReadKey();

Here’s the full Producer listing:

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

namespace Producer
{
  public enum Sector
  {
    Personal,
    Business
  }

  interface ILogger
  {
    void Write(Sector sector, string entry, TraceEventType traceEventType);
  }

  class RabbitLogger : ILogger, IDisposable
  {
    readonly long _clientId;
    readonly IModel _channel;
    bool _disposed;

    public RabbitLogger(IConnection connection, long clientId)
    {
      _clientId = clientId;
      _channel = connection.CreateModel();
      _channel.ExchangeDeclare("direct-exchange-example", ExchangeType.Topic, false, true, null);
    }

    public void Dispose()
    {
      if (!_disposed)
      {
        if (_channel != null && _channel.IsOpen)
        {
          _channel.Close();
        }
      }
      GC.SuppressFinalize(this);
    }

    public void Write(Sector sector, string entry, TraceEventType traceEventType)
    {
      byte[] message = Encoding.UTF8.GetBytes(entry);
      string routingKey = string.Format("{0}.{1}.{2}", _clientId, sector.ToString(), traceEventType.ToString());
      _channel.BasicPublish("topic-exchange-example", routingKey, null, message);
    }

    ~RabbitLogger()
    {
      Dispose();
    }
  }

  class Program
  {
    const long ClientId = 10843;

    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();

      TimeSpan time = TimeSpan.FromSeconds(10);
      var stopwatch = new Stopwatch();
      Console.WriteLine("Running for {0} seconds", time.ToString("ss"));
      stopwatch.Start();

      while (stopwatch.Elapsed < time)
      {
        using (var logger = new RabbitLogger(connection, ClientId))
        {
          Console.Write("Time to complete: {0} seconds\r", (time - stopwatch.Elapsed).ToString("ss"));
          logger.Write(Sector.Personal, "This is an information message", TraceEventType.Information);
          logger.Write(Sector.Business, "This is an warning message", TraceEventType.Warning);
          logger.Write(Sector.Business, "This is an error message", TraceEventType.Error);
          Thread.Sleep(1000);
        }
      }

      connection.Close();
      Console.Write("                             \r");
      Console.WriteLine("Press any key to exit");
      Console.ReadKey();
    }
  }
}

For our Consumer app, we’ll pretty much be using the same code as with our fanout exchange example, but we’ll need to change the exchange type along with the exchange and queue names.  Additionally, we also need to provide a routing key that registers for logs in the Personal sector only.  The messages published by the Producer will be in the form: [client Id].[sector].[log severity], so we can use a routing key of “*.Personal.*” (or alternately “*.Personal.#”).  Here’s the full Consumer listing:

using System;
using System.IO;
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("topic-exchange-example", ExchangeType.Topic, false, true, null);
      channel.QueueDeclare("log", false, false, true, null);
      channel.QueueBind("log", "topic-exchange-example", "*.Personal.*");

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

      while (true)
      {
        try
        {
          var eventArgs = (BasicDeliverEventArgs) consumer.Queue.Dequeue();
          string message = Encoding.UTF8.GetString(eventArgs.Body);
          Console.WriteLine(string.Format("{0} - {1}", eventArgs.RoutingKey, message));
        }
        catch (EndOfStreamException)
        {
          // The consumer was cancelled, the model closed, or the connection went away.
          break;
        }
      }

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

Setting the solution to run both the Producer and Consumer on startup, we should see similar output to the following listings:

Producer

Running for 10 seconds
Time to complete: 06 seconds

Consumer

10843.Personal.Information - This is an information message
10843.Personal.Information - This is an information message
10843.Personal.Information - This is an information message
10843.Personal.Information - This is an information message
10843.Personal.Information - This is an information message
10843.Personal.Information - This is an information message
10843.Personal.Information - This is an information message

This concludes our topic exchange example.  Next time, we’ll walk through an example using the final exchange type: Header Exchanges.

As discussed earlier in the series, the fanout exchange type is useful for facilitating the publish-subscribe pattern. When we publish a message to a fanout exchange, the message is delivered indiscriminately to all bound queues. With the Direct, Topic, and Headers exchange types, a criteria is used by a routing algorithm taking the form of a routing key or a collection of message headers depending on the exchange type in question. A routing key or a collection of message headers may also be specified with the fanout exchange which will be delivered as part of the message’s metadata, but they will not be used as a filter in determining which queue receives a published message.

To demonstrate the fanout exchange, we’ll use a stock ticker example. In the previous example, logs were routed to queues based upon a matching routing key (an empty string in the logging example’s case). In this example, we’d like our messages to be delivered to all bound queues regardless of qualification.

Similar to the previous example, we’ll create a Producer console application which periodically publishes stock quote messages and a Consumer console application which displays the message to the console.

We’ll start our Producer app as before by establishing a connection using the default settings, creating the connection, and creating a channel:

namespace Producer
{
  class Program
  {
    static volatile bool _cancelling;

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

Next, we need to declare an exchange of type “fanout”. We’ll name our new exchange “fanout-exchange-example”:

channel.ExchangeDeclare("fanout-exchange-example", ExchangeType.Fanout, false, true, null);

To publish the stock messages periodically, we’ll call a PublishQuotes() method with the provided channel and run it on a background thread:

var thread = new Thread(() => PublishQuotes(channel));
thread.Start();

Next, we’ll provide a way to exit the application by prompting the user to enter ‘x’ and use a simple Boolean to signal the background thread when to exit:

Console.WriteLine("Press 'x' to exit");
var input = (char) Console.Read();
_cancelling = true;

Lastly, we need to close the channel and connection:

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

For our PublishQuotes() method, well iterate over a set of stock symbols, retrieve the stock information for each symbol, and publish a simple string-based message in the form [symbol]:[price]:

static void PublishQuotes(IModel channel)
{
  while (true)
  {
    if (_cancelling) return;
    IEnumerable quotes = FetchStockQuotes(new[] {"GOOG", "HD", "MCD"});
    foreach (string quote in quotes)
    {
      byte[] message = Encoding.UTF8.GetBytes(quote);
      channel.BasicPublish("fanout-exchange-example", "", null, message);
    }
    Thread.Sleep(5000);
  }
}

To implement the FetchStockQuotes() method, we’ll use the Yahoo Finance API which entails retrieving an XML-based list of stock quotes and parsing out the bit of information we’re interested in for our example:

static IEnumerable<string> FetchStockQuotes(string[] symbols)
{
  var quotes = new List<string>();

  string url = string.Format("http://query.yahooapis.com/v1/public/yql?q=select%20*%20from%20yahoo.finance.quotes%20where%20symbol%20in%20({0})&env=store://datatables.org/alltableswithkeys",
      String.Join("%2C", symbols.Select(s => "%22" + s + "%22")));
  var wc = new WebClient {Proxy = WebRequest.DefaultWebProxy};
  var ms = new MemoryStream(wc.DownloadData(url));
  var reader = new XmlTextReader(ms);
  XDocument doc = XDocument.Load(reader);
  XElement results = doc.Root.Element("results");

  foreach (string symbol in symbols)
  {
    XElement q = results.Elements("quote").First(w => w.Attribute("symbol").Value == symbol);  
    quotes.Add(symbol + ":" + q.Element("AskRealtime").Value);
  }

  return quotes;
}

Here is the complete Producer listing:

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Xml;
using System.Xml.Linq;
using RabbitMQ.Client;

namespace Producer
{
  class Program
  {
    static volatile bool _cancelling;

    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();
      channel.ExchangeDeclare("fanout-exchange-example", ExchangeType.Fanout, false, true, null);

      var thread = new Thread(() => PublishQuotes(channel));
      thread.Start();

      Console.WriteLine("Press 'x' to exit");
      var input = (char) Console.Read();
      _cancelling = true;

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

    static void PublishQuotes(IModel channel)
    {
      while (true)
      {
        if (_cancelling) return;
        IEnumerable quotes = FetchStockQuotes(new[] {"GOOG", "HD", "MCD"});
        foreach (string quote in quotes)
        {
          byte[] message = Encoding.UTF8.GetBytes(quote);
          channel.BasicPublish("fanout-exchange-example", "", null, message);
        }
        Thread.Sleep(5000);
      }
    }


    static IEnumerable<string> FetchStockQuotes(string[] symbols)
    {
      var quotes = new List<string>();

      string url = string.Format("http://query.yahooapis.com/v1/public/yql?q=select%20*%20from%20yahoo.finance.quotes%20where%20symbol%20in%20({0})&env=store://datatables.org/alltableswithkeys",
          String.Join("%2C", symbols.Select(s => "%22" + s + "%22")));
      var wc = new WebClient {Proxy = WebRequest.DefaultWebProxy};
      var ms = new MemoryStream(wc.DownloadData(url));
      var reader = new XmlTextReader(ms);
      XDocument doc = XDocument.Load(reader);
      XElement results = doc.Root.Element("results");

      foreach (string symbol in symbols)
      {
        XElement q = results.Elements("quote").First(w => w.Attribute("symbol").Value == symbol);  
        quotes.Add(symbol + ":" + q.Element("AskRealtime").Value);
      }

      return quotes;
    }
  }
}

Our Consumer application will be similar to the one used in our logging example, but we’ll change the exchange name, queue name, and exchange type and put the processing of the messages within a while loop to continually display our any updates to our stock prices. Here’s the full listing for our Consumer app:

using System;
using System.IO;
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("fanout-exchange-example", ExchangeType.Fanout, false, true, null);
      channel.QueueDeclare("quotes", false, false, true, null);
      channel.QueueBind("quotes", "fanout-exchange-example", "");

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

      while (true)
      {
        try
        {
          var eventArgs = (BasicDeliverEventArgs) consumer.Queue.Dequeue();
          string message = Encoding.UTF8.GetString(eventArgs.Body);
          Console.WriteLine(message);
        }
        catch (EndOfStreamException)
        {
          // The consumer was cancelled, the model closed, or the connection went away.
          break;
        }
      }

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

Setting our solution startup projects to run both the Producer and Consumer apps together, we should see messages similar to the following for the Consumer output:

GOOG:611.62
HD:48.66
MCD:91.06
GOOG:611.58
HD:48.66
MCD:91.06

To show our queue would receive messages published to the fanout exchange regardless of the routing key value, we can change the value of the routing key to “anything”:

channel.QueueBind("quotes", "fanout-exchange-example", "anything");

Running the application again shows the same values:

GOOG:611.62
HD:48.66
MCD:91.06
GOOG:611.58
HD:48.66
MCD:91.06 

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

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.

Exchange Types

Exchanges control the routing of messages to queues.  Each exchange type defines a specific routing algorithm which the server uses to determine which bound queues a published message should be routed to.

RabbitMQ provides four types of exchanges: Direct, Fanout, Topic, and Headers.

Direct Exchanges

The Direct exchange type routes messages with a routing key equal to the routing key declared by the binding queue.  The following illustrates how the direct exchange type works:

The Direct exchange type is useful when you would like to distinguish messages published to the same exchange using a simple string identifier.  This is the type of exchange that was used in our Hello World example.  As discussed in part 3 of our series, every queue is automatically bound to a default exchange using a routing key equal to the queue name.  This default exchange is declared as a Direct exchange.  In our example, the queue named “hello-world-queue” was bound to the default exchange with a routing key of “hello-world-queue”, so publishing a message to the default exchange (identified with an empty string) routed the message to the queue named “hello-world-queue”.

Fanout Exchanges

The Fanout exchange type routes messages to all bound queues indiscriminately.  If a routing key is provided, it will simply be ignored.  The following illustrates how the fanout exchange type works:

The Fanout exchange type is useful for facilitating the publish-subscribe pattern.  When using the fanout exchange type, different queues can be declared to handle messages in different ways.  For instance, a message indicating a customer order has been placed might be received by one queue whose consumers fulfill the order, another whose consumers update a read-only history of orders, and yet another whose consumers record the order for reporting purposes.

Topic Exchanges

The Topic exchange type routes messages to queues whose routing key matches all, or a portion of a routing key.  With topic exchanges, messages are published with routing keys containing a series of words separated by a dot (e.g. “word1.word2.word3”).  Queues binding to a topic exchange supply a matching pattern for the server to use when routing the message.  Patterns may contain an asterisk (“*”) to match a word in a specific position of the routing key, or a hash (“#”) to match zero or more words.  For example, a message published with a routing key of “honda.civic.navy” would match queues bound with “honda.civic.navy”, “*.civic.*”, “honda.#”, or “#”, but would not match “honda.accord.navy”, “honda.accord.silver”, “*.accord.*”, or “ford.#”.  The following illustrates how the fanout exchange type works:

The Topic exchange type is useful for directing messages based on multiple categories (e.g. product type and shipping preference ), or for routing messages originating from multiple sources (e.g. logs containing an application name and severity level).

Headers Exchanges

The Headers exchange type routes messages based upon a matching of message headers to the expected headers specified by the binding queue.  The headers exchange type is similar to the topic exchange type in that more than one criteria can be specified as a filter, but the headers exchange differs in that its criteria is expressed in the message headers as opposed to the routing key, may occur in any order, and may be specified as matching any or all of the specified headers.  The following illustrates how the headers exchange type works:

The Headers exchange type is useful for directing messages which may contain a subset of known criteria where the order is not established and provides a more convenient way of matching based upon the use of complex types as the matching criteria (i.e. a serialized object).

Conclusion

That wraps up our introduction to each of the exchange types.  Next time, we’ll walk through an example which demonstrates declaring a direct exchange explicitly and take a look at the the push API.

Hello World Review

In the previous installment, we walked through creating a Hello World example.  Let’s review the Producer and Consumer code again:

Producer

class Program
{
  static void Main(string[] args)
  {
    var connectionFactory = new ConnectionFactory();
    IConnection connection = connectionFactory.CreateConnection();
    IModel channel = connection.CreateModel();
    channel.QueueDeclare("hello-world-queue", false, false, false, null);
    byte[] message = Encoding.UTF8.GetBytes("Hello, World!");
    channel.BasicPublish(string.Empty, "hello-world-queue", null, message);
    Console.WriteLine("Press any key to exit");
    Console.ReadKey();
    channel.Close();
    connection.Close();
  }
}

Consumer

class Program
{
  static void Main(string[] args)
  {
    var connectionFactory = new ConnectionFactory();
    IConnection connection = connectionFactory.CreateConnection();
    IModel channel = connection.CreateModel();
    channel.QueueDeclare("hello-world-queue", false, false, false, null);
    BasicGetResult result = channel.BasicGet("hello-world-queue", true);
    if (result != null)
    {
      string message = Encoding.UTF8.GetString(result.Body);
      Console.WriteLine(message);
    }
    Console.WriteLine("Press any key to exit");
    Console.ReadKey();
    channel.Close();
    connection.Close();
  }
}

In both the Producer and Consumer projects, we started our Main() method by establishing a Connection to the RabbitMQ server.  We then created a Channel to establish a line of communication with the server and declared a Queue to send and receive our messages.  In our Producer, we called BasicPublish() to place a message on the queue.  In our Consumer, we called BasicGet() to retrieve a message from the queue and print the message to the console.  In both the Producer and Consumer projects, we ended the Main() method by closing the channel and connection to the server.

Establishing the Connection

Let’s first discuss how our example connects to the RabbitMQ server.  In our example, we’re establishing a connection using the default connection settings.

var connectionFactory = new ConnectionFactory();

This assumes we have a RabbitMQ server running on our local development machine.  To connect to this instance, the ConnectionFactory uses a default host name of “localhost” and a default port of 5672.  A user name and password must also be provided to connect to the server.  When not specified, the ConnectionFactory uses a default user name and password of “guest”.  The “guest” user was pre-configured with full administrator rights when we first installed our  instance of RabbitMQ.  The final parameter of interest is the Virtual Host.  RabbitMQ utilizes virtual hosts (i.e. virtual instances of RabbitMQ) to allow multiple projects or teams to manage and secure their own set of queues from within a single RabbitMQ installation.

When we first installed RabbitMQ, a default virtual host named “/” was created.  The ConnectionFactory uses this virtual host when one is not specified.

Specified explicitly, we might have configured these default values as follows:

var connectionFactory = new ConnectionFactory
                          {
                            HostName = "localhost",
                            Port = 5672,
                            UserName = "guest",
                            Password = "guest",
                            VirtualHost = "/"
                          };
IConnection connection = connectionFactory.CreateConnection();

RabbitMQ also supports the AMQP URI Specification which allows these parameters to be specified as a single URI.  Using the URI option, we need to percent-encode the virtual host as “%2f”.  The following is how me might have established our connection using the amqp URI:

var connectionFactory = new ConnectionFactory {Uri = "amqp://guest:guest@localhost:5672/%2f"};
IConnection connection = connectionFactory.CreateConnection();

Creating a Channel

After establishing a connection with the server, the next step we take in both the Producer and Consumer projects is to create a Channel.

IModel channel = connection.CreateModel();

A channel is a light-weight connection used by RabbitMQ to enable multiple communication sessions through a single TCP/IP connection.  Operating systems have a limit to the number of TCP/IP connections that can be opened simultaneously, and creating these connections can be relatively expensive.  Through the use of channels, we are free to create as many virtual connections with the server as we want without incurring this overhead or being limited by the number of available TCP/IP connections.

Channels are intended to be used for sequencing communication with the server such as declaring queues or the sending and receiving of messages.  A single channels should not be used by multiple threads simultaneously. For this reason, it is recommended that channels simply not be shared across threads.

Declaring the Queue

After opening a channel within our connection, the next step we take is to declare a Queue.

channel.QueueDeclare("hello-world-queue", false, false, false, null);

A queue can be thought of as a sort of “mailbox”.  Messages put in the mailbox sit there until a recipient retrieves the mail.

In our example, we named our queue “hello-world-queue”.  The queue name (along with the RabbitMQ host name and port number) serves as a sort of “mailbox address”.  When we publish the encoded string “Hello, World!”, it is placed in the queue named “hello-world-queue’ and awaits until a consumer retrieves and acknowledges the message.

Before discussing the parameters used in our call to QueueDeclare, it will be helpful to take a short detour and discuss how our message actually gets routed from the Producer to the Consumer.

Message Routing

When looking at our example, it appears as though the Producer sends our message directly to the “hello-world-queue”.  What actually occurs, however, is that the message is routed to the “hello-world-queue” through an Exchange.

When using RabbitMQ, messages aren’t actually published to a queue, but rather are published to an exchange which then routes messages to queues.  An exchange can be thought as a “post office”.  Just as we don’t generally put mail directly in a recipient’s mailbox, in RabbitMQ we don’t publish messages directly to a queue.

To receive messages, a queue must first have an exchange binding established.  Part of establishing the binding includes specifying which messages should be routed to the queue.  In most cases, this is achieved by specifying a routing key which serves as a filter for which messages are delivered to which queues.

The following diagram depicts how messages are published in RabbitMQ:

The reason our code looks like it sends messages directly to the queue is due to the fact that we’re taking advantage of a convenience provided by the RabbitMQ server.  Whenever a queue is declared, it is automatically bound to a default exchange with a routing key equal to the queue name.  Messages published to the default exchange with a routing key equal to the value of a declared queue name are routed to the associated queue.

In addition to being implicitly bound to all queues, the default exchange also cannot be unbound, deleted, or explicitly declared through code.  The only operation that is allowed is the publishing of messages.

Queue Parameters

Returning to our discussion of queue declaration, the parameters used in our call to QueueDeclare() creates our queue as a non-durable, non-exclusive, non-auto-deleted queue.

The durability flag pertains to what happens to our queue if the server is restarted.  Queues marked as durable will be recreated upon a server restart, while non-durable queues will not be.

The exclusive flag pertains to whether a queue can be declared by more than one channel.  Since messages are actually published to exchanges, not queues, there may be cases where you only want a single client instance to be capable of declaring a queue bound to a particular exchange (for example, when using queues to facilitate an RPC call).

Lastly, the auto-delete flag pertains to whether the queue is automatically deleted once all consumers are removed.  This flag can be a little misleading since we might suppose setting this to true would cause our “hello-world-queue” to be automatically deleted after our application exists.  In fact, it wouldn’t be.

There are two different approaches to receiving messages: pulling and pushing.  With the pull API, messages are retrieved on demand through a call to BasicGet().  This is the method used in our example.  With the pull API, available messages are delivered to a local queue when a Consumer is bound to the channel.  When setting the auto-delete flag to true, a queue will be deleted once it detects that all consumers have been removed.  While we named the project which receives messages from our queue “Consumer”, the pull API doesn’t actually declare a consumer to retrieve messages.  Therefore, the server is never alerted to the fact that the queue is no longer being used.

Publishing the Message

The last noteworthy point in our Producer project is the parameters to our call to BasicPublish().

channel.BasicPublish(string.Empty, "hello-world-queue", null, message);

The first parameter is the name of the exchange to publish the message to.  An empty string is the designation for the default exchange discussed earlier in the section on message routing.  The second parameter is the routing key.  As previously discussed, all queues are bound to the default exchange using a routing key equal to the value of the bound queue name.  The third parameter is an instance of IBasicProperties.  This class is used to associate a number of properties with the message, such as the message encoding, content-type, and durability.  We’ll take a look at some of the uses of this class further in our series.  Lastly, the fourth parameter is the byte array representing the message body.

Getting the Message

The last noteworthy point in our Consumer project is the parameters to our call to BasicGet().

BasicGetResult result = channel.BasicGet("hello-world-queue", true);

The first parameter is the queue to pull a message from which should be self-explanatory.  The second parameter is a flag which controls the auto-acknowledgement of the message.

When a message is retrieved from a queue with an auto-acknowledge flag set to false, the server holds the message in an unacknowledged state until an acknowledgement or rejection of the message is received on the open channel (note that messages must be accepted or rejected on the same channel they were received on).  This is useful when a consumer needs to ensure the message is processed successfully before it’s fully removed from the server.  If the channel is closed without receiving an acknowledgment, the message will be requeued and delivered to the next consumer (using pull or get).  In the event that a consumer determines it can’t process the message, it can reject the message with a flag indicating whether to requeue the message or not.

Conclusion

That’s it for our Hello World example review.  Next time, we’ll take a look at the four basic exchange types.

Last year, I wrote about my philosophy for dependency management and  how I use NuGet to facilitate dependency management without using the Visual Studio plugin wherein I discuss using the NuGet.exe command line tool to manage .Net dependencies as part of my rake build.  After reading Scott’s post, I got to wondering whether the NuGet.exe command line tool also had the same caching issue and after a bit of testing I discovered that it does.  Since I, with the help of a former colleague, Josh Bush, have evolved the solution I wrote about previously a bit, I thought I’d provide an update to my approach which includes the caching fix.

As discussed in my previous article, I maintain a packages.rb file which serves as a central manifest of all the dependencies used project wide.  Here’s one from a recent project:

packages = [
[ "Machine.Specifications", "0.5.3.0" ],
[ "ExpectedObjects", "1.0.0.2" ],
[ "Moq", "4.0.10827" ],
[ "RabbitMQ.Client", "2.7.1" ],
[ "log4net", "1.2.11" ]
]

configatron.packages = packages

This is sourced by a rakefile which which is used by a task which installs any packages not already installed.

The basic template I use for my rakefile is as follows:

require 'rubygems'
require 'configatron'
 
...

NUGET_CACHE= File.join(ENV['LOCALAPPDATA'], '/NuGet/Cache/') 
FEEDS = ["http://[corporate NuGet Server]:8000", "https://go.microsoft.com/fwlink/?LinkID=206669" ]
require './packages.rb'
 
task :default => ["build:all"]
 
namespace :build do
 
  task :all => [:clean, :dependencies, :compile, :specs, :package] 

  ...

  task :dependencies do
    feeds = FEEDS.map {|x|"-Source " + x }.join(' ')
    configatron.packages.each do | name,version |
      feeds = "-Source #{NUGET_CACHE} " + feeds unless !version
        packageExists = File.directory?("#{LIB_PATH}/#{name}")
        versionInfo="#{LIB_PATH}/#{name}/version.info"
        currentVersion=IO.read(versionInfo) if File.exists?(versionInfo)
        if(!packageExists or !version or !versionInfo or currentVersion != version) then
          versionArg = "-Version #{version}" unless !version
          sh "nuget Install #{name} #{versionArg} -o #{LIB_PATH} #{feeds} -ExcludeVersion" do | ok, results |
            File.open(versionInfo, 'w') {|f| f.write(version) } unless !ok
        end
      end
    end
  end
end

This version defines a NUGET_CACHE variable which points to the local cache.  In the dependencies task, I join all the feeds into a list of Sources for NuGet to check.  I leave out the NUGET_CACHE until I know whether or not a particular package specifies a version number. Otherwise, NuGet would simply check for the latest version which exists within the local cache.

To avoid having to change Visual Studio project references every time I update to a later version of a dependency, I use the –ExcludeVersion option.  This means I can’t rely upon the folder name to determine whether the latest version is already installed, so I’ve introduced a version.info file.  I imagine this is quite a bit faster than allowing NuGet to determine whether the latest version is installed, but I actually do this for a different reason.  If you tell NuGet to install a package into a folder without including the version number as part of the folder and you already have the specified version, it uninstalls and reinstalls the package.  Without checking the presence of the correct version beforehand, NuGet would simply reinstall everything every time.

Granted, this rake task is far nastier than it needs to be.  It should really only have to be this:

  task :dependencies do
    nuget.exe install depedencyManifest.txt –o lib
  end

Where the dependencyManifest file might look a little more like this:

Machine.Specifications 0.5.3.0
ExpectedObjects 1.0.0.2
Moq 4.0.10827
RabbitMQ.Client 2.7.1
log4net 1.2.11

Nevertheless, I’ve been able to coerce the tool into doing what I want for the most part and it all works swimmingly once you get it set up.

Basic Concepts

To being, let’s discuss a few basic concepts. Each of the examples we’ll be working through will have two roles represented: a Producer and a Consumer. A Producer sends messages and a Consumer receives messages.

Messages are basically any blob of bytes you’d like to send. This could be a simple ASCII string, JavaScript Object Notation (JSON), or a binary-serialized object.

Messages are sent to Queues.  A Queue is a First-In-First-Out (FIFO) data structure. You can think of a queue as a sort of pipe where you put messages in one side of the pipe and the messages arrive at the other end of the pipe. 

The following diagram depicts these concepts:

We’ll introduce other concepts further into the series, but that’s the basics. Let’s move on to creating our first example.

Hello, World!

Our first application will be an obligatory “Hello World” example.  We’ll create a Publisher application which sends the string “Hello, World!” to a RabbitMQ queue and a Consumer application which receives the message from the queue and displays it to the console.

For all of our examples, we’ll be using the official RabbitMQ .Net client available here.  This library is also available via NuGet, so if you have the NuGet Package Manager installed you can retrieve it through the “Tools->Library Package Manager” menu item, or if you have the NuGet.exe command line utility then you can issue the following command in the directory you’d like it installed to:

nuget install RabbitMQ.Client

Create the Producer

To start, let’s create a new empty solution named HelloWorldExample (File->New->Project->Other Project Types->Visual Studio Solutions->Blank Solution). Once you have that created, add a new project of type “Console Application” to the solution and name it “Producer”.

Next, add a reference to the RabbitMQ.Client.dll assembly.

The first thing we’ll need to do for our producer is to establish a connection to the RabbitMQ server using a ConnectionFactory:

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

The ConnectionFactory has a number of properties that can be set for our connection. In this example, we’re establishing a connection using the default connection settings which assumes you have the RabbitMQ Windows service running on your local development machine. If you’ve installed it on a different machine then you’ll need to set the Host property of the connectionFactory instance to the DNS name where you’ve installed RabbitMQ.

Next, we need to create a Channel:

IModel channel = connection.CreateModel();

A channel is a light-weight connection which RabbitMQ uses to enable multiple threads of communication over a single TCP/IP socket. Note that the actual type created is RabbitMQ.Client.IModel. In most RabbitMQ client libraries the term channel is used, but for some reason the authors of the .Net client library chose to use the term “Model”. Descriptive, eh? We’ll use the instance name of “channel” to be more descriptive.

Next, we need to create a queue:

channel.QueueDeclare(“hello-world-queue”, false, false, false, null);

This creates a queue on the server named “hello-world-queue” which is non-durable (won’t survive a server restart), is non- exclusive (other channels can connect to the same queue), and is not auto-deleted once it’s no longer being used.  We’ll discuss these parameters in more detail further in our series.

Next, we’ll declare a byte array containing a UTF8-encoded array of bytes from the string “Hello, World!” and use the BasicPublish() method to publish the message to the queue:

byte[] message = Encoding.UTF8.GetBytes("Hello, World!");
channel.BasicPublish(string.Empty, “hello-world-queue”, null, message);

Again, don’t worry about understanding the parameters just yet. We’ll get to that soon enough.

Finally, we’ll prompt the user to press a key to exit the application and close our channel and connection:

Console.WriteLine("Press any key to exit");
Console.ReadKey();
channel.Close();
connection.Close();

Here’s the full Producer listing:

using System.Text;
using RabbitMQ.Client;

namespace Producer
{
  class Program
  {
    static void Main(string[] args)
    {
      var connectionFactory = new ConnectionFactory();
      IConnection connection = connectionFactory.CreateConnection();
      IModel channel = connection.CreateModel();
      channel.QueueDeclare("hello-world-queue", false, false, false, null);
      byte[] message = Encoding.UTF8.GetBytes("Hello, World!");
      channel.BasicPublish(string.Empty, "hello-world-queue", null, message);
      Console.WriteLine("Press any key to exit");
      Console.ReadKey();
      channel.Close();
      connection.Close();
    }
  }
}

Create the Consumer

Next, let’s create our Consumer application. Add a new Console Application to the solution named “Consumer” and add a reference to the RabbitMQ.Client assembly. We’ll start our consumer with the same connection, channel, and queue declarations:

var connectionFactory = new ConnectionFactory();
IConnection connection = connectionFactory.CreateConnection();
IModel channel = connection.CreateModel();
channel.QueueDeclare("hello-world-queue", false, false, false, null);

Next, we’ll use the BasicGet() method to consume the message from the queue “hello-world-queue”:

BasicGetResult result = channel.BasicGet("hello-world-queue", true);

Next, we’ll check to ensure we received a result. If so, we’ll convert the byte array contained within the Body property to a string and display it to the console:

if (result != null)
{
  string message = Encoding.UTF8.GetString(result.Body);
  Console.WriteLine(message);
}

Lastly, we’ll prompt the user to press a key to exit the application and close our channel and connection:

Console.WriteLine("Press any key to exit");
Console.ReadKey();
channel.Close();
connection.Close();

Here’s the full Consumer listing:

using System;
using System.Text;
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.QueueDeclare("hello-world-queue", false, false, false, null);
      BasicGetResult result = channel.BasicGet("hello-world-queue", true);
      if (result != null)
      {
        string message = Encoding.UTF8.GetString(result.Body);
        Console.WriteLine(message);
      }
      Console.WriteLine("Press any key to exit");
      Console.ReadKey();
      channel.Close();
      connection.Close();
    }
  }
}

To see the application in action, start the Publisher application first and then start the Consumer application. If all goes well, you should see the Consumer application print the following:

Hello, World!
Press any key to exit

Congratulations! You’ve just completed your first RabbitMQ application.  Next time, we’ll take a closer look at the concepts used within our Hello World example.

This series will provide a gentle introduction to getting started with RabbitMQ for .Net developers, including a Windows environment installation guide along with an introduction to basic concepts and features through the use of examples in C#.  In this first installment, we’ll cover installation and basic configuration.

Installation

The first thing to know about RabbitMQ installation is that RabbitMQ runs on the Erlang virtual runtime.  “What is Erlang”, you ask, and “Why should I ask our admins to install yet another runtime engine on our servers”?  Erlang is a functional language which places a large emphasis on concurrency and high reliability.  Developed by  Joe Armstrong, Mike Williams, and Robert Virding to support telephony applications at Ericsson, Erlang’s flagship product, the Ericsson AXD301 switch, is purported to have achieved a reliability of nine “9″s.

A popular quote among Erlang adherents is Verding’s “First Rule of Programming”:

“Any sufficiently complicated concurrent program in another language contains an ad hoc informally-specified bug-ridden slow implementation of half of Erlang.” – Robert Verding

Sound like the perfect platform to write a message broker in?  The authors of RabbitMQ thought so too.

With that, let’s get started with the installation.

Step 1: Install Erlang

The first step will be to download and install Erlang for Windows.  You can obtain the latest installer from here (version R15B at the time of this writing) .

After downloading and completing the Erlang installation wizard, you should have a new ERLANG_HOME environment variable set.  If not, you’ll want to set this now so RabbitMQ will be able to locate your installation of Erlang.

Step 2: Install RabbitMQ

Next, download and install the latest version of RabbitMQ for Windows from here (version 2.7.1 at the time of this writing).

Step 3: Install the RabbitMQ Management Plugin

By default, the RabbitMQ Windows installer registers RabbitMQ as a Windows service, so technically we’re all ready to go.  In addition to the command line utilities provided for managing and monitoring our RabbitMQ instance, a Web-based management plugin is also provided with the standard Windows distribution.  The following steps detail how to get the management plugin up and going.

First, from an elevated command prompt, change directory to the sbin folder within the RabbitMQ Server installation directory (e.g. %PROGRAMFILES%\RabbitMQ Server\rabbitmq_server_2.7.1\sbin\).

Next, run the following command to enable the rabbitmq management plugin:

rabbitmq-plugins.bat enable rabbitmq_management 

Lastly, to enable the management plugin we need to reinstall the RabbitMQ service.  Execute the following sequence of commands to reinstall the service:

 
rabbitmq-service.bat stop 
rabbitmq-service.bat install 
rabbitmq-service.bat start 

To verify management plugin is up and running, start your favorite browser and navigate to http://localhost:55672/mgmt/.  If everything went ok, you should see a screen similar to the following:

From here, you’ll be able to configure and monitor your RabbitMQ instance.

That concludes our installation guide.  Next time, we’ll walk through writing our first RabbitMQ C# application.

What Are Closures?

A closure is a pairing of a function along with its referencing environment such that identifiers within the function may refer to variables declared within the referencing environment.

Let’s consider the following example:

var createGreeting = function(greeting) {
    return function(name) {
        document.write(greeting + ', ' + name + '.');
    };
};


helloGreeting = createGreeting("Hello");
howdyGreeting = createGreeting("Howdy");

helloGreeting("John");  // Hello, John.
helloGreeting("Sally"); // Hello, Sally.
howdyGreeting("John");  // Howdy, John.
howdyGreeting("Sally"); // Howdy, Sally.

In this code, a function named createGreeting is defined which returns an anonymous function.  When the anonymous function is executed, it prints a greeting which consists of the outer function’s greeting parameter along with the inner function’s name parameter.  While the greeting parameter is neither a formal parameter  nor a local variable of the inner function (which is to say it is a free variable), it is still resolved when the function executes.  Moreover, the createGreeting function object is no longer in scope and may have even been garbage collected at the point the function executes.  How then does this work?

The inner function is capable of resolving the greeting identifier due to a closure which has been formed for the inner function.  That is to say, the inner function has been paired with its referencing (as opposed to its calling) environment.

Conceptually, we can think of our closure as the marriage between a function and the environment in which it was declared within:

Exactly what this marriage looks like at an implementation level differs depending on the language.  In C# for instance, closures are implemented by the instantiation of a compiler-generated class which encapsulates a delegate and the variables referenced by the delegate from its declaring scope. In JavaScript, the function object itself contains a non-accessible property pointing to an object containing the variables from its declaring scope.  While each implementation differs, both render a function that has access to what its environment looked like at the point it was created.

Let’s move on to examining how JavaScript closures work from a specification perspective.

JavaScript Closures

To understand how closures work in JavaScript, it helps to have a grasp of the underlying concepts set forth within the ECMAScript Language Specification.  I’ll be using Edition 5.1 (ECMA-262) of the specification as the basis for the following discussion.

Execution Contexts

When a JavaScript function is executed, a construct referred to as an Execution Context is created.  The Execution Context is an abstract concept prescribed by the specification to track the execution progress of its associated code.  As an application runs, an initial Global Execution Context is created.  As each new function is created, new Execution Contexts are created which form an Execution Context stack.

There are three primary components prescribed for the Execution Context:

1. The LexicalEnvironment

2. The VariableEnvironment

3. The ThisBinding

Only the LexicalEnvironment and VariableEnvironment components are relevant to the topic of closures, so I’ll exclude discussion of the ThisBinding.  (For information about how the ThisBinding is used, see sections 10.4.1.1, 10.4.2, and 10.4.3 of the ECMAScript Lanauage Specification.)

LexicalEnvironment

The LexicalEnvironment is used to resolve identifier references made by code associated with the execution context.  Conceptually, we can think of the LexicalEnvironment as an object containing the variables and formal parameters declared within the code associated by the current Execution Context.

A LexicalEnvironement itself is comprised of two components: An Environment Record, which is used to store identifier bindings for the Execution Context, and an outer reference which points to a LexicalEnvironment of the declaring Execution Context (which is null in the case of the Global Execution Context’s LexicalEnvironment outer reference).  This forms a chain of LexicalEnvironments, each maintaining a reference to the outer scope’s environment:

VariableEnvironment

The VariableEnvironment is used to record the bindings created by  variables and function declarations within an execution context.  Upon reading that description, you may be thinking: “but I thought the LexicalEnvironment held the bindings for the current execution context”.  Technically, the VariableEnvironment contains the bindings of the variables and function declarations defined within an Execution Context and the LexicalEnvironment is used to resolve the bindings within the Execution Context.  Confused?  The answer to this seemingly incongruous approach is that, in most cases, the LexicalEnvironment and VariableEnvironment are references to the same entity.

The LexicalEnvironment and VariableEnvironment components are actually references to a LexicalEnvironment type.  When the JavaScript interpreter enters the code for a function, a new LexicalEnvironment instance is created and is assigned to both the LexicalEnvironment and VariableEnvironment references.  The variables and function declarations are then recorded to the VariableEnvironment reference.  When the LexicalEnvironment is used to resolve an identifier, any bindings created through the VariableEnvironment reference are available for resolution.

Identifier Resolution

As previously discussed, LexicalEnvironments have an outer reference which, except for the Global Execution Context’s LexicalEnvironment, points to a LexicalEnvironment record of the declaring Execution Context (i.e. a function block’s “parent” scope).  Functions contain an internal scope property (denoted as [[Scope]] by the ECMAScript specification) which is assigned a LexicalEnvironment from the declaring context.  In the case of function expressions (e.g. var doStuff = function() { …}), the [[Scope]] property is assigned to the declaring context’s LexicalEnvironment.  In the case of function declarations (e.g. function doStuff() { … }), the [[Scope]] property is assigned to the declaring context’s VariableEnvironment.  We’ll discuss the reasons for this disparity shortly, but for now let’s just focus on the fact that the function has a [[Scope]] which points to the environment in which is was created.  This is our pairing of a function with its referencing environment, which is to say, this is our closure. 

If we recall our previous conceptualization of a function paired with its environment, the JavaScript version of this conceptualization would look a little like this:

When resolving an identifier, the current LexicalEnvironment is passed to an abstract operation named GetIdentiferReference which checks the current LexicalEnvironment’s Environment Record for the requested identifier and if not found calls itself recursively with the LexicalEnvironment’s outer reference.  Each link of the chain is checked until the top of the chain is reached (the LexicalEnvironment of the Global Context) in which case the binding is resolved or a reference of undefined is returned.

A Distinction Without a Difference … Usually

As mentioned, the LexicalEnvironment and the VariableEnvironement references point to the same LexicalEnvironment instance in most cases.  The reason for maintaining these as separate references is to support the with statement.

The with statement facilitates block scope by using a supplied object as the Environment Record for a newly created LexicalEnvironment.  Consider the following example:

var x = {
    a: 1
};

var doSomething = function() {
    var a = 2;
    
    with(x) {
        console.log(a);
    };
};
doSomething(); // 1

In this code, while the doSomething function assigns the variable a the value of 2, the console.log function logs the value of 1 instead of 2.  What is happening here is that, for the enclosing block of code, the with statement is inserting a new LexicalEnvironment at the front of the chain with an Environment Record set to the object x.  When the code within the with statement executes, the first LexicalEnvironment to be checked will be the one created by the with statement whose Environment Record contains a binding of x with the value of 1.  Once the with statement exits, the LexicalEnvironment of the current Execution Context is restored to its previous state.

According to the ECMAScript specification, function expressions may be declared within a with statement, but not function declarations (though most implementations allow it with varied behavior).  Since function expressions may be declared, their declaration should form a closure based upon the declaring environment’s LexicalEnvironment because it is the LexicalEnvironement which might be changed by a with statement.  This is the reason why Execution Contexts are prescribed  two different LexicalEnvironment references and why function declarations and function expressions differ in which reference their [[Scope]] property is assigned upon function creation.

Conclusion

This concludes our exploration of closures in JavaScript.  While understanding the ECMAScript Language Specification in detail certainly isn’t necessary to have a working knowledge of closures, I find taking a peek under the covers now and again helps to broaden one’s understanding about a language and its concepts.