Messaging with RabbitMQ and .NET review part 5: one way messaging with an event based consumer

Introduction

In the previous post we saw how to process messages from a queue using a receiver we derived from a default basic consumer. We implemented the HandleBasicDeliver function for that purpose. We also discussed two message exchange patterns (MEPs), one-way and and worker queues. The two are practically identical in code but the worker queues MEP implies that we have 2 or more consumers competing for the messages from the queue. That way we can spread out the message load across multiple consumer instances.

In this short post we’ll look at an alternative way to consume messages from a queue in code.

The event based queue consumer

In the previous post we built the OneWayMessageReceiver class which derived from the DefaultBasicConsumer class built into the .NET RabbitMq driver. There is an additional class called EventingBasicConsumer which exposes the message handling functions as events. If you are not sure what events and delegates are in C# you can start here. The end result is the same as we had previously but the syntax is different.

At this point we have a method called ReceiveSingleOneWayMessage in our RabbitMq.OneWayMessage.Receiver console application which is called from the Main method. We’ll now save the channel in a private field and start consuming the messages in an event handler. Here’s the entire code for clarity where I ignored the ReceiveSingleOneWayMessage function:

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace RabbitMq.OneWayMessage.Receiver
{
	class Program
	{
		private static IModel channelForEventing;

		static void Main(string[] args)
		{
			ReceiveMessagesWithEvents();
		}

		private static void ReceiveMessagesWithEvents()
		{
			ConnectionFactory connectionFactory = new ConnectionFactory();

			connectionFactory.Port = 5672;
			connectionFactory.HostName = "localhost";
			connectionFactory.UserName = "accountant";
			connectionFactory.Password = "accountant";
			connectionFactory.VirtualHost = "accounting";

			IConnection connection = connectionFactory.CreateConnection();
			channelForEventing = connection.CreateModel();
			channelForEventing.BasicQos(0, 1, false);
			EventingBasicConsumer eventingBasicConsumer = new EventingBasicConsumer(channelForEventing);
			eventingBasicConsumer.Received += EventingBasicConsumer_Received;
			channelForEventing.BasicConsume("my.first.queue", false, eventingBasicConsumer);
		}

		private static void EventingBasicConsumer_Received(object sender, BasicDeliverEventArgs e)
		{
			IBasicProperties basicProperties = e.BasicProperties;
			Console.WriteLine("Message received by the event based consumer. Check the debug window for details.");
			Debug.WriteLine(string.Concat("Message received from the exchange ", e.Exchange));
			Debug.WriteLine(string.Concat("Content type: ", basicProperties.ContentType));
			Debug.WriteLine(string.Concat("Consumer tag: ", e.ConsumerTag));
			Debug.WriteLine(string.Concat("Delivery tag: ", e.DeliveryTag));
			Debug.WriteLine(string.Concat("Message: ", Encoding.UTF8.GetString(e.Body)));
			channelForEventing.BasicAck(e.DeliveryTag, false);
		}
	}
}

ReceiveMessagesWithEvents starts with the same connection and channel setup code as ReceiveSingleOneWayMessage. We save the channel in the private field to be reused in the event handler EventingBasicConsumer_Received for the acknowledgement. EventingBasicConsumer exposes a number of events of which Received is the most important. We can attach a handler to it which will be fired if there’s a new message in the queue. The body of the event handler function is almost the same as the one we had in the overridden HandleBasicDeliver function of the OneWayMessageReceiver class. The parameters of HandleBasicDeliver are available in the incoming BasicDeliverEventArgs object.

You can set a breakpoint within the event handler and start the receiver application. Then send a message using the publisher we built before. The event handler should be fired with the same effect as in the previous post.

There’s actually an alternative way of registering the event handler using a lambda expression as follows:

private static void ReceiveMessagesWithEvents()
{
	ConnectionFactory connectionFactory = new ConnectionFactory();

	connectionFactory.Port = 5672;
	connectionFactory.HostName = "localhost";
	connectionFactory.UserName = "accountant";
	connectionFactory.Password = "accountant";
	connectionFactory.VirtualHost = "accounting";

	IConnection connection = connectionFactory.CreateConnection();
	IModel channel = connection.CreateModel();
	channel.BasicQos(0, 1, false);
	EventingBasicConsumer eventingBasicConsumer = new EventingBasicConsumer(channel);

	eventingBasicConsumer.Received += (sender, basicDeliveryEventArgs) =>
	{
		IBasicProperties basicProperties = basicDeliveryEventArgs.BasicProperties;
		Console.WriteLine("Message received by the event based consumer. Check the debug window for details.");
		Debug.WriteLine(string.Concat("Message received from the exchange ", basicDeliveryEventArgs.Exchange));
		Debug.WriteLine(string.Concat("Content type: ", basicProperties.ContentType));
		Debug.WriteLine(string.Concat("Consumer tag: ", basicDeliveryEventArgs.ConsumerTag));
		Debug.WriteLine(string.Concat("Delivery tag: ", basicDeliveryEventArgs.DeliveryTag));
		Debug.WriteLine(string.Concat("Message: ", Encoding.UTF8.GetString(basicDeliveryEventArgs.Body)));
		channel.BasicAck(basicDeliveryEventArgs.DeliveryTag, false);
	};
			
	channel.BasicConsume("my.first.queue", false, eventingBasicConsumer);
}

The above solution is based on an anonymous event handler. Moreover, we don’t need to save the channel in a private field anymore. If you don’t understand this lambda syntax then I encourage you to check out the link referenced above which leads you to a series on delegates, events and lambdas.

We’ll explore the fanout exchange type in the next part.

View the list of posts on Messaging here.