6
\$\begingroup\$

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;
                }



            }


}
\$\endgroup\$
  • \$\begingroup\$ Does it work? Why do you think it might not be right? \$\endgroup\$ – svick Feb 26 '14 at 13:20
  • \$\begingroup\$ @svick Yes its work. \$\endgroup\$ – Mahesh Feb 26 '14 at 14:21
5
\$\begingroup\$
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.

\$\endgroup\$
  • \$\begingroup\$ Unfortunately "consumer.Queue.Dequeue(1000, out eventArgs)" do not have any async version. \$\endgroup\$ – Mahesh Mar 11 '14 at 12:06
3
\$\begingroup\$

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!

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.