I am hoping to use EasyNetQ for a consumer task where I will read a batch of messages (say 100) from one or more RabbitMQ queues, and process batches upstream. I don't want to Ack the messages until the upstream process completes , because if my process crashes while the upstream is happening, I'd like the messages to not get dropped.

My test code at the moment looks like (pseudocode - have omitted synchronization detail and so on):

var bus = RabbitHutch.CreateBus("host=localhost");
var consumer = bus.Advanced.Consume(queue, (body, properties, info) => 
    {
        MessageHandler(body, properties, info);
    });

// ...

static void MessageHandler(ReadOnlyMemory<byte> content, MessageProperties props, MessageReceivedInfo info)
{
      messageBatch.Add((content, info));

      if (messageBatch.Count >= 100)
          // trigger another thread to do batch processing
}

The default behaviour of EasyNetQ is that the bus.Advanced.Consume function does an Ack as soon as the handler function returns. Is it possible to turn that off, and if so, then what function can I call to send the Acks later on?

NB. RabbitMQ Consume Messages in Batches and Ack them all at once addresses this question for RabbitMQ.Client, but I would like to know if it's possible in the EasyNetQ client.

I am hoping to use EasyNetQ for a consumer task where I will read a batch of messages (say 100) from one or more RabbitMQ queues, and process batches upstream. I don't want to Ack the messages until the upstream process completes , because if my process crashes while the upstream is happening, I'd like the messages to not get dropped.

My test code at the moment looks like (pseudocode - have omitted synchronization detail and so on):

var bus = RabbitHutch.CreateBus("host=localhost");
var consumer = bus.Advanced.Consume(queue, (body, properties, info) => 
    {
        MessageHandler(body, properties, info);
    });

// ...

static void MessageHandler(ReadOnlyMemory<byte> content, MessageProperties props, MessageReceivedInfo info)
{
      messageBatch.Add((content, info));

      if (messageBatch.Count >= 100)
          // trigger another thread to do batch processing
}

The default behaviour of EasyNetQ is that the bus.Advanced.Consume function does an Ack as soon as the handler function returns. Is it possible to turn that off, and if so, then what function can I call to send the Acks later on?

NB. RabbitMQ Consume Messages in Batches and Ack them all at once addresses this question for RabbitMQ.Client, but I would like to know if it's possible in the EasyNetQ client.

• c#
• rabbitmq
• easynetq

1 Answer 1

From checking the project source, this subject is partially addressed by an undocumented class PullingConsumer:

• Pulling consumer #1073

which includes methods

• PullAsync , AckAsync, RejectAsync (works with 1 message at a time)
• PullBatchAsync, AckBatchAsync, RejectBatchAsync (works with batches of a size that you specify)

The limitations are:

• PullingConsumer can only listen on a single queue at a time (i.e. no wait on multiple queues -- you'd need one of these per queue)

The batch functions don't support partial batches but I think partial batches could be handled by doing PullAsync and then when something is received, try a PullBatchAsync(n-1, ... with a very short timeout.