RabbitMQ Consumer

Question

I am playing with RabbitMQ and .NET and found myself duplicating the connection initialization logic in all of the consumers.

Created the Consumer in order to avoid that and add some tracing. It's far away from anything usable, but wanted to hear ideas/opinions/alternatives.

public class Consumer
{
  private static readonly Logger _logger = LogManager.GetCurrentClassLogger();
  public event EventHandler<CanonicalModel> OnReceive;
  private IModel _exchange;

  public void Init(string incoming = "#")
  {
    var factory = new ConnectionFactory() { HostName = "localhost" };
    using (var connection = factory.CreateConnection())
      using (_exchange = connection.CreateModel())
      {
        _exchange.ExchangeDeclare("exchange_name", "topic");
        var queueName = _exchange.QueueDeclare().QueueName;
        _exchange.QueueBind(queueName, "queue_name", incoming);

        var consumer = new QueueingBasicConsumer(_exchange);
        _exchange.BasicConsume(queueName, true, consumer);

        while (true)
        {
          var ea = consumer.Queue.Dequeue();
          var model = new CanonicalModel(ea.Body); 

          _logger.Info(new { Time = DateTime.Now, Type = "IN", Value = "..." });

          var startTime = DateTime.Now;
          if (OnReceive != null)
          {
            OnReceive(this, model);
          }

          _logger.Info(new { Time = DateTime.Now, Type = "CONTROL", Value = DateTime.Now.Subtract(startTime).Milliseconds });
        }
      }
  }

  public void Publish(string routingKey, CanonicalModel model)
  {
    var json = JsonConvert.SerializeObject(model);
    var message = Encoding.UTF8.GetBytes(json);

    _logger.Info(new { Time = DateTime.Now, Type = "OUT", Value = "..." });
    _exchange.BasicPublish("exchange_name", routingKey, null, message);
  }
}

Console application which uses a consumer.

class ConsoleConsumer
{
  private const string SERVICENAME = "CONSUMER_FOO";
  private const string INCOMING = "foo";
  private const string OK = "success";
  private const string FAIL = "fail";
  private static Consumer _consumer;

  static void Main(string[] args)
  {

    Console.WriteLine(" [Running] {0} ...", SERVICENAME);

    _consumer = new Consumer();
    _consumer.OnReceive += Process;
    _consumer.Init(INCOMING);

  }

  public static void Process(object sender, CanonicalModel model)
  {
    var result = true;
    var outRoutingKey = result ? OK : FAIL;

    _consumer.Publish(outRoutingKey, model);
  }
}

Answer 1

I like the consumer as a consumer and not as an initializer--which means I would split all the initialization logic out into a separate class that takes in parameters and returns your IModel. That way you can pass the IModel directly into the consumer already initialized.

You also might want to consider using the observer pattern and the .NET IObserver/IObservable interfaces if you are looking for a layer of abstraction.