Method enqueue

dramatiq/brokers/rabbitmq.py:355–425

Enqueue a message. Parameters: message(Message): The message to enqueue. delay(int): The minimum amount of time, in milliseconds, to delay the message by. Raises: ConnectionClosed: If the underlying channel or connection has bee

(self, message: Message, *, delay: Optional[int] = None)

        )

    def enqueue(self, message: Message, *, delay: Optional[int] = None) -> Message:
        """Enqueue a message.

        Parameters:
          message(Message): The message to enqueue.
          delay(int): The minimum amount of time, in milliseconds, to
            delay the message by.

        Raises:
          ConnectionClosed: If the underlying channel or connection
            has been closed.
        """
        canonical_queue_name = message.queue_name
        queue_name = canonical_queue_name

        if delay is not None:
            queue_name = dq_name(queue_name)
            message_eta = current_millis() + delay
            message = message.copy(
                queue_name=queue_name,
                options={
                    "eta": message_eta,
                },
            )

        attempts = 0
        while True:
            try:
                self.declare_queue(canonical_queue_name, ensure=True)
                self.logger.debug("Enqueueing message %r on queue %r.", message.message_id, queue_name)
                self.emit_before("enqueue", message, delay)
                self.channel.basic_publish(
                    exchange="",
                    routing_key=queue_name,
                    body=message.encode(),
                    properties=pika.BasicProperties(
                        delivery_mode=2,
                        priority=message.options.get("broker_priority"),
                    ),
                    # mandatory flag ensures UnroutableError is raised if message could not be routed to a queue,
                    # but it only works when confirm_delivery is turned on, so only set it when that is the case.
                    # https://www.rabbitmq.com/docs/publishers#unroutable
                    mandatory=self.confirm_delivery,
                )
                self.emit_after("enqueue", message, delay)
                return message

            except (
                pika.exceptions.AMQPConnectionError,
                pika.exceptions.AMQPChannelError,
            ) as e:
                # Delete the channel and the connection so that the
                # next caller/attempt may initiate new ones of each.
                del self.connection

                # If the queue disappears, add it to the set of pending queues
                # so that it can be redeclared on retry or the next time
                # a message is enqueued.