def _on_message(self, delivery_frame, properties, body, consumer):
        """
        Callback when a message is received from the server.

        This method wraps a user-registered callback for message delivery. It
        decodes the message body, determines the message schema to validate the
        message with, and validates the message before passing it on to the
        user callback.

        This also handles acking, nacking, and rejecting messages based on
        exceptions raised by the consumer callback. For detailed documentation
        on the user-provided callback, see the user guide on consuming.

        Args:
            delivery_frame (pika.spec.Deliver): The delivery frame which
                includes details about the message like content encoding and
                its delivery tag.
            properties (pika.spec.BasicProperties): The message properties like
                the message headers.
            body (bytes): The message payload.
            consumer (dict): A dictionary describing the consumer of the message.

        Returns:
            Deferred: fired when the message has been handled.
        """
        _legacy_twisted_log.msg(
            "Message arrived with delivery tag {tag} for {consumer}",
            tag=delivery_frame.delivery_tag,
            consumer=consumer,
            logLevel=logging.DEBUG,
        )
        try:
            message = get_message(delivery_frame.routing_key, properties, body)
            message.queue = consumer.queue
        except ValidationError:
            _legacy_twisted_log.msg(
                "Message id {msgid} did not pass validation; ignoring message",
                msgid=properties.message_id,
                logLevel=logging.WARNING,
            )
            yield consumer.channel.basic_nack(
                delivery_tag=delivery_frame.delivery_tag, requeue=False
            )
            return

        try:
            _legacy_twisted_log.msg(
                "Consuming message from topic {topic!r} (id {msgid})",
                topic=message.topic,
                msgid=properties.message_id,
            )
            yield defer.maybeDeferred(consumer.callback, message)
        except Nack:
            _legacy_twisted_log.msg(
                "Returning message id {msgid} to the queue",
                msgid=properties.message_id,
                logLevel=logging.WARNING,
            )
            yield consumer.channel.basic_nack(
                delivery_tag=delivery_frame.delivery_tag, requeue=True
            )
        except Drop:
            _legacy_twisted_log.msg(
                "Consumer requested message id {msgid} be dropped",
                msgid=properties.message_id,
                logLevel=logging.WARNING,
            )
            yield consumer.channel.basic_nack(
                delivery_tag=delivery_frame.delivery_tag, requeue=False
            )
        except HaltConsumer:
            _legacy_twisted_log.msg(
                "Consumer indicated it wishes consumption to halt, shutting down"
            )
            yield consumer.channel.basic_ack(delivery_tag=delivery_frame.delivery_tag)
            yield self.cancel(consumer.queue)
        except Exception:
            _legacy_twisted_log.msg(
                "Received unexpected exception from consumer {c}",
                c=consumer,
                logLevel=logging.ERROR,
            )
            yield consumer.channel.basic_nack(
                delivery_tag=0, multiple=True, requeue=True
            )
            yield self.cancel(consumer.queue)
        else:
            yield consumer.channel.basic_ack(delivery_tag=delivery_frame.delivery_tag)