feat: add ensure_consumer kwarg to before/after_declare_queue to control whether a consumer should be created on a RabbitMQ, then disable consumer thread creation upon message delivery

[jenstroeger]

Following the discussion Issue with RabbitMQ Broker: sending a message to the queue creates a consumer thread locally, this draft illustrates the proposed change.

@Bogdanp I also noticed that there’s no

self.emit_before("declare_delay_queue", delayed_name)

callback when the DQ is declared — is that intentional?

[Bogdanp]

Thanks, but, unfortunately, this would be a breaking change for anyone already implementing one of these hooks, so I don't think this can work (and I don't think the problem warrants complicating the code to add arity checks to work around the breakage). Until we can come up with a better solution, I think the way to go is that I proposed in that discussion. A potential alternative would be to allow some level of worker configuration -- eg. to allow a user to extend the worker middleware in some way, but I haven't had time to think about the implications of allowing that.

[jenstroeger]

Thanks, but, unfortunately, this would be a breaking change for anyone already implementing one of these hooks, so I don't think this can work (and I don't think the problem warrants complicating the code to add arity checks to work around the breakage).

That’s why I used kwargs with defaults so that existing code continues to work.

Until we can come up with a better solution, I think the way to go is that I proposed in that discussion.

I agree, though I’m not enthusiastic about either one approach… 😉

A potential alternative would be to allow some level of worker configuration -- eg. to allow a user to extend the worker middleware in some way, but I haven't had time to think about the implications of allowing that.

And that leads back to exposing maybe too much of Dramatiq internals. Hmm. Maybe let’s sit on it for a few days?

Also, what about the other question above, the missing invocation of the before_declare_delay_queue callback?