RabbitMQ missing queue re-declaring does not work

[LincolnPuzey]

Relates to #556

Checklist

• Does your title concisely summarize the problem?
• Did you include a minimal, reproducible example?
• What OS are you using?
• What version of Python are you using?
• What version of Dramatiq are you using?
• What Broker are you using?
• What did you do?
• What did you expect would happen?

What OS are you using?

Linux mint 22.3

What version of Python are you using?

3.10.19

What version of Dramatiq are you using?

master branch

What Broker are you using?

RabbitMQ - server version 4.2.2

What did you do?

Using the following minimal app:

import dramatiq.brokers.rabbitmq

broker = dramatiq.brokers.rabbitmq.RabbitmqBroker(confirm_delivery=True)
dramatiq.set_broker(broker)


@dramatiq.actor
def foo():
    pass

When deleting a Queue, dramatiq does not detect this and does not recreate it.
This happens both from the publisher and consumer side.

On the consumer side, if you;

1. Start a Worker to ensure the queues are created
2. Delete the queue while the Worker is running

The worker then constantly loops with the error:

 [dramatiq.worker.ConsumerThread(default)] [CRITICAL] Consumer encountered a connection error: (404, "NOT_FOUND - no queue 'default' in vhost '/'")

and the queue is not re-declared.

On the publisher side, if you

1. Send one message to ensure queue is created
2. Delete the queue
3. Send another message

Sending the second messages silently fails (!) to send the message, because Dramatiq does not use the mandatory flag.

What did you expect would happen?

Not sure, but #556 implies that the queue should be re-declared.

[LincolnPuzey]

From what I can tell, the change in #556 does not work,
so dramatiq will currently never re-declare queues after creating them for the first time, if they subsequently get deleted while dramatiq is running.
(and likely never had this functionality?).

Given that fixing this brings RabbitMQ behavior in-line with redis:
(where the hash/list data structures are automatically created if they don't exist)

dramatiq/dramatiq/brokers/redis/dispatch.lua

Lines 145 to 150 in 05a1b53

	if command == "enqueue" then
	local message_id = ARGS[1]
	local message_data = ARGS[2]
	redis.call("hset", queue_messages, message_id, message_data)
	redis.call("rpush", queue_full_name, message_id)

and because its useful functionality, I think we should fix this.