BasicAck when processing messages asynchronously
<p>I'm trying to set up a RabbitMQ messaging queue so that I can send a message to start a long running process and also be able to send a message to cancel that long running process if needed. So I started out with an <code>EventingBasicConsumer</code> and did something like this in my <code>Recieved</code> handler:</p>
<pre><code>if (startProcess) 
{
    // start a long running process
}
else if (cancelProcess)
{
    // cancel the currently running process
}
channel.BasicAck(ea.DeliveryTag, false);
</code></pre>
<p>And this doesn't work because the <code>EventingBasicConsumer</code> isn't multithreaded and can only handle one message at a time. So it can't handle the cancel message until it's finished with the long running process (at which point, there's no point, obviously). So next I tried this:</p>
<pre><code>if (startProcess) 
{
    Task.Run(() =&gt; {
        // start a long running process
    }
}
else if (cancelProcess)
{
    // cancel the currently running process
}
channel.BasicAck(ea.DeliveryTag, false);
</code></pre>
<p>And this works. I can now cancel the long running process...but, I'm acknowledging the request to run the long running process immediately, rather than after it's completed. This means that if the long running process was to crash, the message has already been removed. So this would require the original sender to keep track and have the receiver have to send messages back to say it's done and it all gets a bit complicated.</p>
<p>So I thought maybe I could change <code>EventingBasicConsumer</code> to just always fire its <code>Received</code> event on a new thread. So I created something like this:</p>
<pre><code>public class AsyncRabbitConsumer : DefaultBasicConsumer
{
    // code all the same as EventingBasicConsumer except this bit:
    public override void HandleBasicDeliver(string consumerTag,
        ulong deliveryTag,
        bool redelivered,
        string exchange,
        string routingKey,
        IBasicProperties properties,
        byte[] body)
    {
        base.HandleBasicDeliver(consumerTag,
            deliveryTag,
            redelivered,
            exchange,
            routingKey,
            properties,
            body);
        if (Received != null)
        {
            var args = new BasicDeliverEventArgs(consumerTag,
                    deliveryTag,
                    redelivered,
                    exchange,
                    routingKey,
                    properties,
                    body);

            Task.Run(() =&gt;
            {
                Received(this, args);
            });
        }
    }
}
</code></pre>
<p>Now in my first snippet of code, I can have it process the cancel message while the long running process is still running <em>and</em> the long running process won't Ack and delete it's message until it's actually finished (or cancelled). So that should be great...except when I cancel I get this:</p>
<blockquote>
<p>An exception of type 'RabbitMQ.Client.Exceptions.AlreadyClosedException' occurred in RabbitMQ.Client.dll but was not handled in user code</p>
<p>Additional information: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text=&quot;PRECONDITION_FAILED - unknown delivery tag 3&quot;, classId=60, methodId=80, cause=</p>
</blockquote>
<p>From the <code>channel.BasicAck</code> step of what appears to be the thread that started the long running process. So what's going on here? I think the acknowledgements (for the cancel message first and then the long running process message) are getting crossed here. Is there any decent way to straighten this out? Or am I barking up the wrong tree?</p>
<p>It's probably worth noting that cancelling the long running process is not instantaneous. It will cancel at the next convenient point, so it's almost certain that the cancel message will finish processing before the long running process has ended.</p>