How do I configure PurgeOnStartup for a queue created through AddConsumer()?

Question

Some context to prevent the XY problem: I'm working on a system in which a message is sent every X minutes to run a job on a pod in a service. When all the pods are unhealthy or restarting, these messages can build up in the queue. Then when the pods restart they start consuming all the messages at once. This can cause problems, since it's important the job is not run more often than once every X minutes.

A solution to this appears to be the PurgeOnStartup flag. It would purge the queue when a pod restarts and then it can wait for the next job message to come along. I'm having some trouble adding the PurgeOnStartup to our configuration.

Currently, we add consumers to the configuration like this:

configurator.AddConsumer<SendTemplatedEmailMessage2Consumer>(c => c.UseConcurrentMessageLimit(1));

Which is then used by this code from a central nuget package:

            configurator.UsingRabbitMq((context, cfg) =>
                {
                    if (settings.RabbitmqDelayedMessageExchangeEnabled)
                    {
                        cfg.UseConsumeFilter(typeof(HealthyConsumerFilter<>), context);
                    }
                    cfg.Host(settings.GetUri(), host =>
                    {
                        host.Username(settings.UserName);
                        host.Password(settings.Password);
                    });
                    cfg.ConfigureEndpoints(context);

The second part is setup which we use in all our services, so I'd prefer not to change it. I also only want to add it to this one queue. However, the AddConsumer() configurator doesn't seem to allow for PurgeOnStartup to be set. I've gotten lost amidst filters and pipespecifications a bit and it appears not to be the way. Is it possible to configure a PurgeOnStartup on a queue created through AddConsumer()? If not, how do I add it to a single consumer/queue through the UsingRabbitMq config?

Answer 1

Well, it seems like you want to set the PurgeOnStartup flag for a certain queue that has been created through the AddConsumer method in RabbitMQ. Unfortunately, the AddConsumer method itself does not directly support such configuration of the PurgeOnStartup flag.

For that, you may have to go the other way around. You can attempt to build the consumer manually and supply it with the required configurations - in other words, let the PurgeOnStartup flag yourself - without passing through the AddConsumer method.

Here's how you could manage to do it:

cfg.ReceiveEndpoint("your_queue_name", endpoint =>
{
    endpoint.UseConcurrentMessageLimit(1); // Your existing configuration
    
    // Set PurgeOnStartup for this specific queue
    endpoint.PurgeOnStartup = true; // Set this flag to purge on startup

    endpoint.Consumer<SendTemplatedEmailMessage2Consumer>(context);
});

That way you can create the consumer for the specific queue (your_queue_name should be replaced with your actual queue name) and set the PurgeOnStartup flag for that queue while keeping the rest of your configuration intact.