Listen to multiple RabbitMQ queue by task and process the message

Question

Single app which listen to multiple RabbitMQ queue and process the message. This is working fine but not sure this implementation is right one or I am missing something.

Implementation is inspired from this answer https://stackoverflow.com/a/21847234/37571

//Message subscriber implementation
public class AuditSubscriber : IMessageSubscriber
 {
 public IList<string> SubscribedRouteKeys
 {
 get { return new List<string>() 
 {
 "*.inceitive.attested.*"
 }; 
 }
 }
 public async Task<bool> Process(Core.MessageInfo MessageItem)
 {
 //Start new task to process the message
 bool _ProcessedResult = await Task<bool>.Factory.StartNew(() => MessageProcesser(MessageItem), CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
 return _ProcessedResult;
 }
 protected bool MessageProcesser(MessageInfo MessageItem)
 {
 Thread.Sleep(1000); //Acthual work
 return true;
 }
 }
public class RabbitMQMessageConsumer : AbstractRabbitMQClient, IMessageConsumer
{
 //Message consumer method, which will initiate number of tasks based upon the available subscriber. 
 public void Consume(CancellationToken token)
 {
 //Start Rabbit MQ connection
 StartConnection(_ConnectionFactory.Get());
 List<Task> tasks = new List<Task>();
 foreach (SubscriberType subscriberType in (SubscriberType[])Enum.GetValues(typeof(SubscriberType)))
 {
 //Start listeing to all queues based upon the number of subscriber type availbale in the system 
 Task task = Task.Factory.StartNew(() => ConsumeMessage(subscriberType, token), CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
 tasks.Add(task);
 }
 Task.WhenAll(tasks);
 }
 //Listen to queue
 async Task ConsumeMessage(SubscriberType subscriberType, CancellationToken token)
 {
 try
 {
 //Get message subscriber which will process the message
 IMessageSubscriber _MessageSubscriber = _MessageSubscriberFactory.GetMessageSubscriber(subscriberType);
 using (IModel _ConsumerChannel = _Connection.CreateModel())
 {
 _ConsumerChannel.ExchangeDeclare(_ExchangeProperties.Name, _ExchangeProperties.Type, _ExchangeProperties.Durable);
 string _QueueName = Enum.GetName(typeof(SubscriberType), subscriberType);
 _ConsumerChannel.QueueDeclare(_QueueName, _QueueProperties.Durable, _QueueProperties.Exclusive, _QueueProperties.AutoDelete, _QueueProperties.Arguments);
 foreach (string routeKey in _MessageSubscriber.SubscribedRouteKeys)
 {
 _ConsumerChannel.QueueBind(_QueueName, _ExchangeProperties.Name, routeKey);
 }
 var consumer = new QueueingBasicConsumer(_ConsumerChannel);
 _ConsumerChannel.BasicConsume(_QueueName, false, consumer);
 //Infinite loop to listen the queueu
 while (true)
 {
 if (token.IsCancellationRequested)
 {
 break;
 }
 try
 { 
 BasicDeliverEventArgs eventArgs;
 //Get meesage or time out 
 if (consumer.Queue.Dequeue(1000, out eventArgs))
 {
 if (eventArgs != null)
 {
 MessageInfo _MessageItem = ByteArrayToMessageInfo(eventArgs.Body);
 //Message process by async method
 var messageProcesser = _MessageSubscriber.Process(_MessageItem);
 //Wait for result 
 bool _MessageProcessed = await messageProcesser;
 if (_MessageProcessed)
 {
 _ConsumerChannel.BasicAck(eventArgs.DeliveryTag, false);
 }
 else
 {
 _ConsumerChannel.BasicNack(eventArgs.DeliveryTag, false, true);
 }
 }
 else
 {
 //connnection is dead
 }
 }
 }
 catch (EndOfStreamException ex)
 {
 Console.WriteLine(ex.Message);
 throw;
 }
 }
 }
 }
 catch (Exception ex)
 {
 Console.WriteLine(ex.Message);
 //TODO: Restart the task again
 throw;
 }
 }
}

• svick: Does it work? Why do you think it might not be right?
• Mahesh: @svick Yes its work.

Answer 1

public async Task<bool> Process(Core.MessageInfo MessageItem)
{
 //Start new task to process the message
 bool _ProcessedResult = await Task<bool>.Factory.StartNew(() => MessageProcesser(MessageItem), CancellationToken.None, TaskCreationOptions.LongRunning, TaskScheduler.Default);
 return _ProcessedResult;
}

• That's a weird name for a local variable. The common naming is camelCase (e.g. processedResult). Underscores are sometimes used for fields, but certainly not for locals.

• The variable is actually unnecessary, you can just directly return the expression.

• If all awaits in a method are return awaits, then you don't need await at all, just return the Task directly, after removing async from the signature.

• Are you sure TaskCreationOptions.LongRunning is appropriate here? Its practical (and undocumented) effect is that it creates a new Thread to execute the Task. If you don't need that, just use Task.Run()

With all those changes the method would look like this:

public Task<bool> Process(Core.MessageInfo MessageItem)
{
 return Task.Run(() => MessageProcesser(MessageItem));
}

---

There are many empty lines in your code that I think are unnecessary. Empty lines are useful, but I think the way you're using them (after { or between two }) is just wasting space. And multiple empty lines are usually not useful either.

---

Task.WhenAll(tasks);

WhenAll() returns a Task that represents waiting for all the passed-in Tasks, so ignoring its return value like this doesn't make any sense. Ideally, you should await the returned Task, but for that you need async. And async void methods shouldn't be used. So, if you need to wait for all the Tasks here and you can't use await, you will have to block the thread by using Task.WaitAll(tasks).

---

consumer.Queue.Dequeue(1000, out eventArgs)

This looks like a blocking method. Isn't there an asynchronous version available? If there is, you should probably use that instead.

• Mahesh: Unfortunately "consumer.Queue.Dequeue(1000, out eventArgs)" do not have any async version.

Answer 2

Very new to all the async await stuff so I could be completely off the mark here but I don't really follow why you are running the MessageProcessor in another new task.

//Wait for result 
bool _MessageProcessed = await messageProcesser;

You are already consuming messages in an asynchronous task. So it seems to me that your call to await on the messageProcessor is just going to block an existing asynchronous task. So you might as well just do that work in the current asynchronous task rather than submitting a new task? There is no caller of ConsumeMessage that benefits from this additional await.

Like I said - very new to async await, so please 'school' me if I'm missing the point!