CODEWITHSUNDEEP
c#.netrabbitmq
Pyrejkee
Pyrejkee

Reputation: 337

RabbitMQ doesn't remove message after it was consumed

I got a simple producer and consumer.

I want to mannualy manage the process of deleting messages, so I set autoAck = false on consumer side. The problem is once a message has been consumed - it's not removed from queu and stil marked as ready (when with set autoAck = true it's removed immediately).

using RabbitMQ.Client;
using System.Text;

var factory = new ConnectionFactory();
factory.UserName = "myuser";
factory.Password = "mypassword";
factory.HostName = "myhost";
factory.Port = myport;
factory.VirtualHost = "myvhost";
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queue: "notifications",
                         durable: true,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);
    var properties = channel.CreateBasicProperties();
    properties.Persistent = true;

    string message = "Hello World!";
    var body = Encoding.UTF8.GetBytes(message);

    channel.BasicPublish(exchange: "",
                         routingKey: "notifications",
                         basicProperties: properties,
                         body: body);
    Console.WriteLine(" [x] Sent {0}", message);
}

Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();

Here is my consumer

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var factory = new ConnectionFactory();
factory.UserName = "myuser";
factory.Password = "mypassword";
factory.HostName = "myhost";
factory.Port = myport;
factory.VirtualHost = "myvhost";

using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queue: "notifications",
                         durable: true,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        Console.WriteLine(" [x] Received {0}", message);
        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
    };
    channel.BasicConsume(queue: "notifications",
                         autoAck: false,
                         consumer: consumer);
}

Console.WriteLine(" Press [enter] to exit.");
Console.ReadLine();

Upvotes: 1

Views: 2009

Answers (1)

Luke Bakken
Luke Bakken

Reputation: 9627

Your consumer code should not dispose the connection and channel instance until exiting. If you dispose early, like your original code, there will be nothing to call back to:

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var factory = new ConnectionFactory();
factory.UserName = "myuser";
factory.Password = "mypassword";
factory.HostName = "myhost";
factory.Port = myport;
factory.VirtualHost = "myvhost";

using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.QueueDeclare(queue: "notifications",
                         durable: true,
                         exclusive: false,
                         autoDelete: false,
                         arguments: null);

    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();
        var message = Encoding.UTF8.GetString(body);
        Console.WriteLine(" [x] Received {0}", message);
        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
    };
    channel.BasicConsume(queue: "notifications",
                         autoAck: false,
                         consumer: consumer);

    Console.WriteLine(" Press [enter] to exit.");
    Console.ReadLine();
}

Upvotes: 3

Related Questions