How to correctly finish RabbitMq consumer and wait to RabbitMq consumer threads (library consumer callbacks) after close connection and model?

This is really bug in RabbitMQ.Client (v5.1.2). Source code of ConsumerWorkService.cs:

namespace RabbitMQ.Client
{
    public class ConsumerWorkService
    {
        ...

        class WorkPool
        {
            readonly ConcurrentQueue<Action> actions;
            readonly AutoResetEvent messageArrived;
            readonly TimeSpan waitTime;
            readonly CancellationTokenSource tokenSource;
            readonly string name;

            public WorkPool(IModel model)
            {
                name = model.ToString();
                actions = new ConcurrentQueue<Action>();
                messageArrived = new AutoResetEvent(false);
                waitTime = TimeSpan.FromMilliseconds(100);
                tokenSource = new CancellationTokenSource();
            }

            public void Start()
            {
#if NETFX_CORE
                System.Threading.Tasks.Task.Factory.StartNew(Loop, System.Threading.Tasks.TaskCreationOptions.LongRunning);
#else
                var thread = new Thread(Loop)
                {
                    Name = "WorkPool-" + name,
                    IsBackground = true
                };
                thread.Start();
#endif
            }

            public void Enqueue(Action action)
            {
                actions.Enqueue(action);
                messageArrived.Set();
            }

            void Loop()
            {
                while (tokenSource.IsCancellationRequested == false)
                {
                    Action action;
                    while (actions.TryDequeue(out action))
                    {
                        try
                        {
                            action();
                        }
                        catch (Exception)
                        {
                        }
                    }

                    messageArrived.WaitOne(waitTime);
                }
            }

            public void Stop()
            {
                tokenSource.Cancel();
            }
        }
    }
}
        

As we see there is no any waitings of thread var thread = new Thread(Loop). So really event RabbitMQ.Client.Events.EventingBasicConsumer::Received can be fired anytime even when there is no consumer or connection for a long time which closed long time ago till internal library queue will empty. As I supposed(( :

Action action;
while (actions.TryDequeue(out action))
{
    try
    {
        action();
    }
    catch (Exception)
    {
    }
}

So IModel::Close() will set only CancelationToken without join of thread and there is need some workaround for this Bug.

Just to confirm @Alexander's findings, this issue is still present in v6.2.2 of the .Net Client and the issue is also present with the event driven callback consumer.

I've found that:

• The 'Received' callback registered with both the EventingBasicConsumer and AsyncEventingBasicConsumer will be invoked for up to the prefetchCount setting of the BasicQos setup on the channel, after the connection is closed.
• In manual Ack mode, the call to BasicAck will throw a RabbitMQ.Client.Exceptions.AlreadyClosedException if the connection is closed, and the message won't be acked from the queue, however, the callback will continue to process subsequent messages. This could lead to idempotence problems with receiving the same message more than once during a disconnect.

This may be another good reason to ensure that the prefetchCount is set to a finite and sane value (in addition to e.g. Memory considerations of unbounded prefetch counts).

Finally, if I deliberately close the Connection unexpectedly (e.g. during testing), I found that I needed to explicitly detach my consumer Received handler AND explicitly call consumerChannel.Close(); (i.e. IModel.Close) before I could create a new connection (the Rabbit client would tend to 'hang').

• The connection.ConnectionShutdown event doesn't seem to fire reliably if I have set a ConsumerDispatchConcurrency > 1 with the synchronous EventingBasicConsumer
• The Shutdown event on AsyncEventingBasicConsumer doesn't fire either, if I have ConsumerDispatchConcurrency > 1` when using the async consumer.
• However, I did find that the ModelShutdown event on the IModel / channel fires reliably for sync / async and concurrency > 1.

consumerChannel.ModelShutdown += (sender, args) =>
{
    consumer.Received -= handler; // e.g. AsyncEventingBasicConsumer
    consumerChannel.Close(); // IModel
};