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)
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`