BasicConsume using EventingBasicConsumer with timeout

[lesio999]

my case is that I'm listening on dedicated queue for message form dedicated external process. I do not know if the process that process is running or even if it is running is publishing to my que. I cannot wait for that message forever. It would be nice if I can handle this case via timeout on consume.

if there is no such a thing like timeout on consume, I'm open for some other suggestions. I tried implement ManualResetEvent which works fine with low workload, but with heavy one introducing big performance degradation.

[lukebakken]

if there is no such a thing like timeout on consume

There isn't.

Please create a git repository that I can clone, compile, and run to see what you have tried to do with ManualResetEvent. Hosting it on GitHub would be idea because I can fork it to provide suggestions.

[lesio999]

I cannot share the whole project or so, this is not simple exec app, but this is the code for manualreset, if you spot anything strage....

_var response = new BlockingCollection<string>();

var consumerChannel = Connection.CreateModel();
consumerChannel.BasicQos(0, 1, false);

var consumerTag = "";
var deliveryTag = 0UL;

using (var signal = new ManualResetEvent(false))
{
    var consumer = new EventingBasicConsumer(consumerChannel);
    consumer.Received += (model, ea) =>
    {
        var body = ea.Body.ToArray();

        var messageBody = Encoding.UTF8.GetString(body);
        if (ea.BasicProperties.CorrelationId == correlationId)
        {
            deliveryTag = ea.DeliveryTag;

            response.Add(messageBody);
            response.CompleteAdding();

            consumerChannel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
        }

        signal.Set();
    };

    consumerTag = consumerChannel.BasicConsume(consumer: consumer, queue: replyQueueName, autoAck: false);

    // wait until message is received or timeout reached
    bool timeout = !signal.WaitOne(TimeSpan.FromMilliseconds(_consumeTimeoutMilliseconds));
    if (timeout)
    {
        // cancel subscription
        consumerChannel.BasicCancel(consumerTag);
   }

    consumerChannel.Close();
};

var r = response.Take();

[lukebakken]

Basically, since you only ever consume one message per iteration of this code, you're polling the queue for messages. Is that a correct reading of your process?

Also, please note that if the incoming message does not have the correct correlation ID, it won't ever be acked or nacked. This probably isn't what you want.

[lukebakken]

@lesio999 - please see this code, which demonstrates how to coordinate between a publisher, consumer, and worker. I guessed at what you're trying to accomplish:

https://github.com/lukebakken/rabbitmq-dotnet-client-1816/blob/main/Program.cs

[ansd]

if there is no such a thing like timeout on consume, I'm open for some other suggestions.

Such a thing exists in AMQP 1.0, see Synchronous Get With Timeout in https://www.rabbitmq.com/blog/2024/09/02/amqp-flow-control#drain
I would expect that https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client would offer such a method?

[lukebakken]

Yes, AMQP 0.9.1 does offer basic.get which we try to tell people to avoid 😉

[lesio999]

I know, I tried it as well, it did not perform to our needs

[michaelklishin]

@lesio999 this is a monitoring problem, one way or another. Polling (please don't use polling, it's 2025) or a long-running consumer that automatically shuts down both can work but it's up to your application to monitor the state of the queue or the external process (if it's feasible at all), not RabbitMQ's and nor of this client library.

[lesio999]

as I said "I tried it", did not say, it was good few years ago - did not work for me back then, for sure will not play with it today.
Please treat this as closed - I have my answer.

[lesio999]

thanks a lot