code: def enqueue(self, message, *, delay=None):
          
          queue_name = message.queue_name
          properties = pika.BasicProperties(
              delivery_mode=2,
              priority=message.options.get("broker_priority"),
          )
  
          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 = 1
          while True:
              try:
                  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=properties,
                  )
                  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.channel
                  del self.connection
  
                  attempts += 1
                  if attempts > MAX_ENQUEUE_ATTEMPTS:
                      raise ConnectionClosed(e) from None
  
                  self.logger.debug(
                      "Retrying enqueue due to closed connection. [%d/%d]",
                      attempts, MAX_ENQUEUE_ATTEMPTS,
                  )
docs: 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.