docs: def rpc_reply(self,
                    body,
                    properties=None,
                    exchange=None,
                    reply_to=None,
                    connection=None):
          """Reply to the message that is currently being processed.
  
          .. versionadded:: 4.0.0
  
          If the exchange is not specified, the exchange of the message that is
          currently being processed by the Consumer will be used.
  
          If ``reply_to`` is not provided, it will attempt to use the
          ``reply_to`` property of the message that is currently being processed
          by the consumer. If both are not set, a :exc:`ValueError` is raised.
  
          If any of the following message properties are not provided, they will
          automatically be assigned:
  
            - ``app_id``
            - ``correlation_id``
            - ``message_id``
            - ``timestamp``
  
          The ``correlation_id`` will only be automatically assigned if the
          original message provided a ``message_id``.
  
          If the connection that the message is being published on has publisher
          confirmations enabled, a :py:class:`~tornado.concurrent.Future` is
          returned.
  
          :param body: The message body
          :type body: :class:`bytes` or :class:`str`
          :param properties: The AMQP properties to use for the reply message
          :type properties: :class:`dict` or :data:`None`
          :param exchange: The exchange to publish to. Defaults to the exchange
              of the message that is currently being processed by the Consumer.
          :type exchange: :class:`str` or :data:`None`
          :param reply_to: The routing key to send the reply to. Defaults to the
              reply_to property of the message that is currently being processed
              by the Consumer. If neither are set, a :exc:`ValueError` is
              raised.
          :type reply_to: class:`str` or :data:`None`
          :param connection: The connection to use. If it is not
              specified, the channel that the message was delivered on is used.
          :type connection: :class:`str`
          :rtype: :class:`tornado.concurrent.Future` or :data:`None`
          :raises: :exc:`ValueError`
  
          """
          if reply_to is None and self.reply_to is None:
              raise ValueError('Missing reply_to routing key')
          properties = properties or {}
          if not properties.get('app_id'):
              properties['app_id'] = self.name
          if not properties.get('correlation_id') and self.message_id:
              properties['correlation_id'] = self.message_id
          if not properties.get('message_id'):
              properties['message_id'] = str(uuid.uuid4())
          if not properties.get('timestamp'):
              properties['timestamp'] = int(time.time())
          return self.publish_message(exchange or self.exchange, reply_to
                                      or self.reply_to, properties, body,
                                      connection)