In my last post “A trivial message loop using RabbitMQ in C#.NET” I demonstrated a simple concept of a circular message queue using RabbitMQ. I chose not to complicate the post with explaining some fundamentals of RabbitMQ leaving that for this post.
As of writing the version of RabbitMQ I am using is 2.4.1.
Some of the concepts I introduced in the last post were, and I’ll break them down in a hopefully a logical flow as steps to take.
- IConnection
- IModel
- ExchangeDeclare()
- QueueHelper.EnsureQueue()
- Subscription
- SimpleRpcClient
- SimpleRpcClient.Cast()
- Your own SimpleRpcServer
- HandleSimpleCast
First if you’re looking to get setup on Windows head on over to the RabbitMQ official documentation and follow the steps.
Also before you continue reading I urge you to go have a look at Mike Hadlow’s EasyNeqQ API and accompanying blog posts.
Some other good blog posts I have stumbled upon after starting to write my version of this come from Simon Dixon, at the time of me writing this he had 1, 2, 3 part series
On to my take, I’ll be explaining RabbitMQ along the lines of how my simple demonstration application functions, as a reminder it’s available on BitBucket – https://bitbucket.org/NickJosevski/rabbitloop
Step 1 – The Connection
I am using ConnectionFactory with a local machine address to create the connection the client and server will use to build the next required component.
IConnection connection = new ConnectionFactory
{
Address = "127.0.0.1" //localhost
}.CreateConnection();
Step 2 – The Model
Using the connection create a model this is used to configure the rest of details on how communication will take place between the client and server.
IModel model = connection.CreateModel();
Step 3 – ExchangeDeclare()
Declaring an exchange is the concept of setting up a named path for communication, the second parameter offers options on how messages are to be delivered, in this example I’m sticking with a simple choice of ‘Direct’
var exchange = "my.exchange";
model.ExchangeDeclare(exchange, ExchangeType.Direct);
Step 4 – QueueHelper.EnsureQueue()
This is my own helper extension method to help perform a few tasks; creating a queueId, using that queueId to declare a queue on the model, and then bind the queue and exchange together, finally returning the newly created queueName.
public static class QueueHelper
{
public static string EnsureQueue(this IModel ch, String exchangeName)
{
var queueId = String.Format("{0}.reply", exchangeName);
var queueName = ch.QueueDeclare(queueId, false, false, false, null);
ch.QueueBind(queueName, exchangeName, "", null);
return queueName;
}
}
Step 5 – Subscription
The subscription is supplied to the server logic and is created using the model and the queueName
var sub = Subscription(model, queueName);
Step 6 – SimpleRpcClient
Creating the client is just as simple, and we’re ready to send a message.
var client = new SimpleRpcClient(model, queueName);
Step 7 – SimpleRpcClient.Cast()
The Cast() method on the SimpleRpcClient is the asynchronous way to send a message, I want to keep this post more straight forward so I am excluding the message class and how to serialize it but those steps are very basic.
var myMsg = new SerializableClassYouHaveCreated().Serialize();
rpcClient.Cast(new BasicProperties(), myMsg);
Step 8 – NicksSimpleRpcServer
This is where things start to get a little more involved, but it’s relatively straight forward. In this approach I create my own RpcServer that extends SimpleRpcServer.
public class NicksSimpleRpcServer : SimpleRpcServer
{
private readonly SimpleRpcClient rpcClient;
public NicksSimpleRpcServer(Subscription subscription, SimpleRpcClient client)
: base(subscription)
{
rpcClient = client;
}
}
Step 9 – HandleSimpleCast
All I do now is override the method that handles asynchronous messages (it also needs to de-serialize the content form byte[] body). Just to follow along from my sample application the server also has a client private member so it can continue to forward a modified message on wards.
//as an override method inside NicksSimpleRpcServer:
public override void HandleSimpleCast(Boolean isRedelivered, IBasicProperties requestProperties, byte[] body)
{
//deal with incoming message, create new message to forward
rpcClient.Cast(new BasicProperties(), msgToForward);
}
This is where our story comes to an end… As a final step we need to get our server running…
Using the Task Parallel Library create a Task that kicks off another method of the RpcServer – MainLoop(). MainLoop simply sits there (blocks) and accepts incoming messages, each time an asynchronous arrives HandleSimpleCast will fire and the message will be processed.
var server = new NicksSimpleRpcServer(sub, client);
new Task(server.MainLoop).Start();
Hope this is easy to follow.
