CODEWITHSUNDEEP
c#async-awaitrabbitmqtask
advapi
advapi

Reputation: 3907

RabbitMq Handle Received message in async way

I'm using RabbitMq to process messages I receive on a bus. I was wondering if there's a better way to process the message I receive (maybe using async/await pattern)

Here's a snippet of my code

connection = connectionFactory.CreateConnection();

channel = connection.CreateModel();

channel.QueueDeclare(queue: Constants.RabbitListeningQueue,durable: false,exclusive: false,autoDelete: false,arguments: null);
channel.QueueDeclare(queue: Constants.RabbitMqRequestInsertedQueue,durable: false,exclusive: false,autoDelete: false,arguments: null);

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
    log.Debug($"[x] Received message :{ea}");
    var body = ea.Body;
    var message = Encoding.UTF8.GetString(body);

    var dynamicObject = JObject.Parse(message);

    queueMessageHandler.HandleMessage(dynamicObject);
};

The queueMessageHandler implementation is as follows

public class QueueMessageHandler : IQueueMessageHandler
{
    private readonly IImportNucleoManager importNucleoManager;

    public QueueMessageHandler(IImportNucleoManager importNucleoManager)
    {
        this.importNucleoManager = importNucleoManager;
    }

    public void HandleMessage(dynamic message)
    {
        switch ((string)message.Type)
        {
            case "T1":
            {
                importNucleoManager.Process(message);
                break;
            }
            case "T3":
                importNucleoManager.ProceedToInsertStep(message);
                break;
        }
    }
}

I was wondering (since the T1/T3 events take a long time to process) should they be Task and so even the HandleMessage should be HandleMessageAsync? In this case, I also have to pass an async void which is not a best practice as I know

Upvotes: 2

Views: 5205

Answers (1)

Creyke
Creyke

Reputation: 2097

static async Task Main(string[] args)
{
    var connectionFactory = new ConnectionFactory(DispatchConsumersAsync = true);
    var connection = connectionFactory.CreateConnection();
    var channel = connection.CreateModel();

    var consumer = new AsyncEventingBasicConsumer(channel);
    consumer.Received += Consumer_Received;
}

static async Task Consumer_Received(object sender, BasicDeliverEventArgs @event)
{
    await DoSomethingAsync();
}

Upvotes: 9

Related Questions