Add configurable timeout for break_on_empty

[eandersson]

Replace the hardcoded 1.0s empty-queue wait in build_inbound_messages with a configurable empty_timeout param (default unchanged). process_data_events now uses empty_timeout=None so start_consuming returns as soon as the queue drains, restoring 3.0.x consume responsiveness without dropping in-flight messages.

[eandersson]

@tichi73 would you be open to reviewing this change?

[tichi73]

Thanks for the follow-up @eandersson — making the grace configurable and having process_data_events pass empty_timeout=None cleanly restores the 3.0.x responsiveness while keeping a knob for direct build_inbound_messages callers. I validated the branch against RabbitMQ 4.3.1 (8 runs): large-message start_consuming() is back to a ~1450 MB/s median (vs ~112 on stock 3.1.x), a no_ack post-cancel test loses nothing, and unit tests + flake8 are green.

One thing I'd suggest fixing — a post-cancel in-flight message can be dropped. Moving if not self.consumer_tags: break ahead of the break_on_empty/empty guard means build_inbound_messages breaks while _inbound still holds a lone Basic.Deliver whose header/body haven't been read yet — dropping a message the broker had already started delivering. Deterministic repro (no broker; identical build_inbound_messages(break_on_empty=True) call): Deliver enqueued → consumer cancelled → header+body arrive:

this branch (empty_timeout): delivered per run: [0, 0, 0, 0, 0]   # dropped
with the fix below:          delivered per run: [1, 1, 1, 1, 1]

The window is microscopic on loopback (which is why a no_ack post-cancel throughput test shows no loss), but it widens under network latency/load, and it's on the normal start_consuming path (the break fires regardless of empty_timeout). The one-line fix is to break only when the queue is also empty:

-                if not self.consumer_tags:
+                if not self.consumer_tags and not self._inbound:
                     break

Repro: https://gist.github.com/tichi73/b753d22734163e5e40d14aa563fdc03c

A few smaller things — entirely your call:

• start_consuming() with no consumer now returns silently rather than raising AMQPChannelError('no consumer callback defined') (the :raises still in the docstring).
• With the current placement, build_inbound_messages also exits on an empty queue with no consumer even when break_on_empty=False, so break_on_empty no longer fully controls exit-on-empty.
• The empty_timeout docstring says only None exits immediately, but any falsy value (e.g. 0) does too — and it isn't validated (non-numeric → TypeError, negative → immediate break).
• The new empty_timeout parameter has no dedicated test yet (None/0 → immediate; a custom value → waits that long).

I prototyped fixes + tests for all of these on a branch (validated: the drop repro goes 0/5, throughput unchanged, 426 unit tests green) — happy to open a PR if useful. The in-flight drop is the only one I'd consider important; the rest are at your discretion.

[eandersson]

Thanks for the follow-up @eandersson — making the grace configurable and having process_data_events pass empty_timeout=None cleanly restores the 3.0.x responsiveness while keeping a knob for direct build_inbound_messages callers. I validated the branch against RabbitMQ 4.3.1 (8 runs): large-message start_consuming() is back to a ~1450 MB/s median (vs ~112 on stock 3.1.x), a no_ack post-cancel test loses nothing, and unit tests + flake8 are green.

One thing I'd suggest fixing — a post-cancel in-flight message can be dropped. Moving if not self.consumer_tags: break ahead of the break_on_empty/empty guard means build_inbound_messages breaks while _inbound still holds a lone Basic.Deliver whose header/body haven't been read yet — dropping a message the broker had already started delivering. Deterministic repro (no broker; identical build_inbound_messages(break_on_empty=True) call): Deliver enqueued → consumer cancelled → header+body arrive:

this branch (empty_timeout): delivered per run: [0, 0, 0, 0, 0]   # dropped
with the fix below:          delivered per run: [1, 1, 1, 1, 1]

The window is microscopic on loopback (which is why a no_ack post-cancel throughput test shows no loss), but it widens under network latency/load, and it's on the normal start_consuming path (the break fires regardless of empty_timeout). The one-line fix is to break only when the queue is also empty:

-                if not self.consumer_tags:
+                if not self.consumer_tags and not self._inbound:
                     break

Repro: https://gist.github.com/tichi73/b753d22734163e5e40d14aa563fdc03c

A few smaller things — entirely your call:

• start_consuming() with no consumer now returns silently rather than raising AMQPChannelError('no consumer callback defined') (the :raises still in the docstring).
• With the current placement, build_inbound_messages also exits on an empty queue with no consumer even when break_on_empty=False, so break_on_empty no longer fully controls exit-on-empty.
• The empty_timeout docstring says only None exits immediately, but any falsy value (e.g. 0) does too — and it isn't validated (non-numeric → TypeError, negative → immediate break).
• The new empty_timeout parameter has no dedicated test yet (None/0 → immediate; a custom value → waits that long).

I prototyped fixes + tests for all of these on a branch (validated: the drop repro goes 0/5, throughput unchanged, 426 unit tests green) — happy to open a PR if useful. The in-flight drop is the only one I'd consider important; the rest are at your discretion.

Thanks for the thorough review! I've fixed the most critical issues and left the rest to keep this PR manageable. If it looks good to you and you have time, a follow-up on the remaining points after it's merged would be much appreciated.

[tichi73]

Confirmed on 0f47f0d: the in-flight drop is gone (repro now 1/1 kept, 0/5 dropped) and start_consuming raises again — moving the consumer_tags break inside the guard also resolves the break_on_empty=False point. LGTM. Once this merges I'll send the follow-up for the empty_timeout docstring/validation (+ a focused test for the param). Thanks!

[eandersson]

Confirmed on 0f47f0d: the in-flight drop is gone (repro now 1/1 kept, 0/5 dropped) and start_consuming raises again — moving the consumer_tags break inside the guard also resolves the break_on_empty=False point. LGTM. Once this merges I'll send the follow-up for the empty_timeout docstring/validation (+ a focused test for the param). Thanks!

Thanks again!