<div dir="ltr">Thanks Simon,<div><br></div><div>I've tried global as true and false for BasicQoS and see the same behavior. I've got some quick example code below that demonstrates the issue I'm having and a zip archive of the project is available <a href="https://www.dropbox.com/s/x7q9e6ijuxzp9zn/RabbitQoSTest.zip">here</a>. I captured network traffic from WireShark and it seems that calling BasicQoS() and increasing the prefetchCount, the qos change gets sent to the server, and the server replies with a qos-ok, but the server doesn't send more messages if the new prefetch count is greater than the outstanding message count for the channel. It's not until I Ack the first message that the second one is sent.</div>
<div><br></div><div>It seems to me that an Ack from a client is causing the server to look at the queue being consumed and do something like:</div><div><br></div><div>if(queue.has-messages && channel.unacked-messages < channel.prefetch-count)</div>
<div> send-messages-to-channel-until-unacked-equals-prefetch-count(channel)</div><div><br></div><div>For the behavior I'd like to see, it seems like a call to basic.qos would require the same logic.</div><div><br>
</div><div>Below is a snippet of my example code. ActionConsumer is a simple IBasicConsumer that invokes the action passed to it's constructor during HandleBasicDeliver. It also uses a simple extension method to List<T> called WaitFor() which waits for the list to contain the specified number of items. I've added comments based on my network capture of this code being executed.</div>
<div><br></div><div>Thanks for any help. Right now, I've got a pretty ugly workaround, which is that instead of increasing prefetch count and calling basicqos, I just do a basicget, but this is definitely hacky and makes me poll using basicget() on another thread, etc...</div>
<div><br></div><div>Thanks.</div><div><br></div><div><br></div><div><div><div>var consumer = new ActionConsumer(Model, message =></div><div> Task.Run(() =></div><div> {</div><div> Log.DebugFormat("Received({0})", message);</div>
<div> messages.Add(Encoding.UTF8.GetString(message.Body));</div>
<div> if (messages.Count == 1)</div><div> {</div><div> //If this is the first message, send a second, set qos to 2 and wait for second message to arrive</div><div> Send("World"); //This message gets sent immediately</div>
<div> Log.Debug("BasicQoS(0, 2, false)");</div><div> Model.BasicQos(0, 2, false); //QoS is sent to server immediately, prefetch shows as 2 in management console</div><div> Log.Debug("WaitFor(2, 5000)");</div>
<div> messages.WaitFor(2, 5000); //While waiting, 2nd message will not be received</div><div> Log.DebugFormat("BasicAck({0}, false)", message.DeliveryTag);</div><div> Model.BasicAck(message.DeliveryTag, false); //As soon as Ack is sent, 2nd message is transmitted to client</div>
<div> messages.WaitFor(2, 2500);</div><div> Log.Debug("Finished task 1 processing");</div><div> }</div><div> }));</div><div><br></div><div>//Start consuming</div><div>Model.BasicConsume(queue, false, consumer);</div>
<div> </div><div>//Send starter message</div><div>Send("Hello");</div><div>var ret = messages.WaitFor(2, 15000);</div></div></div></div><div class="gmail_extra"><br><br><div class="gmail_quote">On Mon, May 12, 2014 at 4:08 AM, Simon MacMullen <span dir="ltr"><<a href="mailto:simon@rabbitmq.com" target="_blank">simon@rabbitmq.com</a>></span> wrote:<br>
<blockquote class="gmail_quote" style="margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex">On 11/05/2014 04:55, Jon Stelly wrote:<br>
<blockquote class="gmail_quote" style="margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex">
I have a sort of odd scenario. I want to process 1 message at a time so<br>
I call basic qos with a prefetch of 1 and call BasicConsume(), but<br>
sometimes my processing will end in a long blocking task. When that<br>
happens, I would like to increase the qos prefetch to 2 to trigger the<br>
next message to be received and processed. The issue is that I don't<br>
want to ack the first message until after the long blocking process<br>
completes successfully.<br>
<br>
If I call BasicQoS before acking the first message, the 2nd message<br>
isn't received until after the long running task completes and the ack<br>
is sent. Can anyone explain what's going on and possibly suggest an<br>
alternative?<br>
</blockquote>
<br>
You are running 3.3.x? The new semantics for consumer prefetch mean that setting basicQos after consuming will only define a prefetch for *new* consumers. You could work around this by setting global=true.<br>
<br>
See <a href="http://www.rabbitmq.com/consumer-prefetch.html" target="_blank">http://www.rabbitmq.com/<u></u>consumer-prefetch.html</a> for more information.<br>
<br>
I am starting to wonder if the global=false qos should set the prefetch for all consumers on the channel; you wouldn't be able to have one channel with multiple consumers with different prefetch, but this is an area that seems to be unintuitive...<br>
<br>
Cheers, Simon<br>
</blockquote></div><br></div>