def acknowledge_cumulative()

in pulsar/__init__.py [0:0]

• 7 lines of code
• 3 McCabe index (conditional complexity)

    def acknowledge_cumulative(self, message: Union[Message, MessageId, _pulsar.Message, _pulsar.MessageId]):
        """
        Acknowledge the reception of all the messages in the stream up to (and
        including) the provided message.

        This method will block until an acknowledgement is sent to the broker.
        After that, the messages will not be re-delivered to this consumer.

        Parameters
        ----------

        message:
            The received message or message id.

        Raises
        ------
        CumulativeAcknowledgementNotAllowedError
            if the consumer type is ConsumerType.KeyShared or ConsumerType.Shared
        """
        if isinstance(message, Message):
            self._consumer.acknowledge_cumulative(message._message)
        elif isinstance(message, MessageId):
            self._consumer.acknowledge_cumulative(message._msg_id)
        else:
            self._consumer.acknowledge_cumulative(message)