(
        self, stream_names: List[str], consumer_group: str, expected_events: set
    )
Reclaim batches of messages that other consumers in the group failed to acknowledge within a timeout.

        The timeout period is specified by the `acknowledgement_timeout` option.
Reclaim batches of messages that other consumers in the group failed to acknowledge within a timeout.
Reclaim batches of messages that other consumers in the group failed to acknowledge within a timeout .
async def _reclaim_lost_messages(
        self, stream_names: List[str], consumer_group: str, expected_events: set
    ) -> AsyncGenerator[List[EventMessage], None]:
        """Reclaim batches of messages that other consumers in the group failed to acknowledge within a timeout.

        The timeout period is specified by the `acknowledgement_timeout` option.
        """
        with await self.connection_manager() as redis:
            for stream in stream_names:

                old_messages = True
                reclaim_from = None

                # Keep pulling reclaimable messages from Redis until there are none left
                while old_messages:
                    # reclaim_from keeps track of where we are up to in our fetching
                    # of messages
                    if not reclaim_from:
                        # This is our first iteration, so fetch from the start of time
                        reclaim_from = "-"
                    else:
                        # This is a subsequent iteration. XPENDING's 'start' parameter is inclusive,
                        # so we need to add one to the reclaim_from value to ensure we don't get a message
                        # we've already seen
                        reclaim_from = redis_stream_id_add_one(reclaim_from)

                    # Fetch the next batch of messages
                    try:
                        old_messages = await redis.xpending(
                            stream, consumer_group, reclaim_from, "+", count=self.reclaim_batch_size
                        )
                    except ReplyError as e:
                        if "NOGROUP" in str(e):
                            # Group or consumer doesn't exist yet, so stop processing for this loop.
                            break
                        else:
                            raise

                    timeout = self.acknowledgement_timeout * 1000
                    event_messages = []

                    # Try to claim each messages
                    for (
                        message_id,
                        consumer_name,
                        ms_since_last_delivery,
                        num_deliveries,
                    ) in old_messages:
                        message_id = decode(message_id, "utf8")
                        consumer_name = decode(consumer_name, "utf8")
                        reclaim_from = message_id

                        # This 'if' is not strictly required as the subsequent call to xclaim
                        # will honor the timeout parameter. However, using this if here allows
                        # for more sane logging from the point of view of the user. Without it
                        # we would report that we were trying to claim messages which were
                        # clearly not timed out yet.
                        if ms_since_last_delivery > timeout:
                            logger.info(
                                L(
                                    "Found timed out event {} in stream {}. Abandoned by {}."
                                    " Attempting to reclaim...",
                                    Bold(message_id),
                                    Bold(stream),
                                    Bold(consumer_name),
                                )
                            )

                            # *Try* to claim the message...
                            result = await redis.xclaim(
                                stream, consumer_group, self.consumer_name, int(timeout), message_id
                            )

                            # Parse each message we managed to claim
                            for _, fields in result:
                                # Note that sometimes we will claim a message and it will be 'nil'.
                                # In this case the result will be (None, {}). We therefore do not
                                # rely on the values we get back from XCLAIM.
                                # I suspect this may happen if a stream has been trimmed, thereby causing
                                # un-acknowledged messages to be deleted from Redis.
                                try:
                                    event_message = self._fields_to_message(
                                        fields,
                                        expected_events,
                                        stream=stream,
                                        native_id=message_id,
                                        consumer_group=consumer_group,
                                    )
                                except (NoopMessage, IgnoreMessage):
                                    # This listener doesn't need to care about this message, so acknowledge
                                    # it and move on with our lives
                                    logger.debug(
                                        f"Ignoring NOOP event with ID {message_id} discovered"
                                        " during event reclaiming"
                                    )
                                    await redis.xack(stream, consumer_group, message_id)
                                    continue

                                logger.debug(
                                    LBullets(
                                        L(
                                            "⬅ Reclaimed timed out event {} on stream {}. Abandoned"
                                            " by {}.",
                                            Bold(message_id),
                                            Bold(stream),
                                            Bold(consumer_name),
                                        ),
                                        items=dict(
                                            **event_message.get_metadata(),
                                            kwargs=event_message.get_kwargs(),
                                        ),
                                    )
                                )
                                event_messages.append(event_message)

                    # And yield our batch of messages
                    if event_messages:
                        yield event_messages
async def _reclaim_lost_messages ( self , stream_names : List [ str ] , consumer_group : str , expected_events : set ) -> AsyncGenerator [ List [ EventMessage ] , None ] : with await self . connection_manager ( ) as redis : for stream in stream_names : old_messages = True reclaim_from = None # Keep pulling reclaimable messages from Redis until there are none left while old_messages : # reclaim_from keeps track of where we are up to in our fetching # of messages if not reclaim_from : # This is our first iteration, so fetch from the start of time reclaim_from = "-" else : # This is a subsequent iteration. XPENDING's 'start' parameter is inclusive, # so we need to add one to the reclaim_from value to ensure we don't get a message # we've already seen reclaim_from = redis_stream_id_add_one ( reclaim_from ) # Fetch the next batch of messages try : old_messages = await redis . xpending ( stream , consumer_group , reclaim_from , "+" , count = self . reclaim_batch_size ) except ReplyError as e : if "NOGROUP" in str ( e ) : # Group or consumer doesn't exist yet, so stop processing for this loop. break else : raise timeout = self . acknowledgement_timeout * 1000 event_messages = [ ] # Try to claim each messages for ( message_id , consumer_name , ms_since_last_delivery , num_deliveries , ) in old_messages : message_id = decode ( message_id , "utf8" ) consumer_name = decode ( consumer_name , "utf8" ) reclaim_from = message_id # This 'if' is not strictly required as the subsequent call to xclaim # will honor the timeout parameter. However, using this if here allows # for more sane logging from the point of view of the user. Without it # we would report that we were trying to claim messages which were # clearly not timed out yet. if ms_since_last_delivery > timeout : logger . info ( L ( "Found timed out event {} in stream {}. Abandoned by {}." " Attempting to reclaim..." , Bold ( message_id ) , Bold ( stream ) , Bold ( consumer_name ) , ) ) # *Try* to claim the message... result = await redis . xclaim ( stream , consumer_group , self . consumer_name , int ( timeout ) , message_id ) # Parse each message we managed to claim for _ , fields in result : # Note that sometimes we will claim a message and it will be 'nil'. # In this case the result will be (None, {}). We therefore do not # rely on the values we get back from XCLAIM. # I suspect this may happen if a stream has been trimmed, thereby causing # un-acknowledged messages to be deleted from Redis. try : event_message = self . _fields_to_message ( fields , expected_events , stream = stream , native_id = message_id , consumer_group = consumer_group , ) except ( NoopMessage , IgnoreMessage ) : # This listener doesn't need to care about this message, so acknowledge # it and move on with our lives logger . debug ( f"Ignoring NOOP event with ID {message_id} discovered" " during event reclaiming" ) await redis . xack ( stream , consumer_group , message_id ) continue logger . debug ( LBullets ( L ( "⬅ Reclaimed timed out event {} on stream {}. Abandoned" " by {}." , Bold ( message_id ) , Bold ( stream ) , Bold ( consumer_name ) , ) , items = dict ( * * event_message . get_metadata ( ) , kwargs = event_message . get_kwargs ( ) , ) , ) ) event_messages . append ( event_message ) # And yield our batch of messages if event_messages : yield event_messages