(self, *event_messages: RedisEventMessage)
Acknowledge that a message has been successfully processed
Acknowledge that a message has been successfully processed
Acknowledge that a message has been successfully processed
async def acknowledge(self, *event_messages: RedisEventMessage):
        """Acknowledge that a message has been successfully processed"""
        with await self.connection_manager() as redis:
            p = redis.pipeline()
            for event_message in event_messages:
                p.xack(event_message.stream, event_message.consumer_group, event_message.native_id)
                logging.debug(
                    f"Preparing to acknowledge message {event_message.id} (Native ID:"
                    f" {event_message.native_id})"
                )

            logger.debug(
                f"Batch acknowledging successful processing of {len(event_messages)} message."
            )
            await p.execute()
async def acknowledge ( self , * event_messages : RedisEventMessage ) : with await self . connection_manager ( ) as redis : p = redis . pipeline ( ) for event_message in event_messages : p . xack ( event_message . stream , event_message . consumer_group , event_message . native_id ) logging . debug ( f"Preparing to acknowledge message {event_message.id} (Native ID:" f" {event_message.native_id})" ) logger . debug ( f"Batch acknowledging successful processing of {len(event_messages)} message." ) await p . execute ( )