(
        self,
        api_name: str,
        event_name: str,
        listener_name: str,
        include_pending: bool = True,
        max_count: Optional[int] = 100,
    )
return lag
How far behind is the given listener for the given api/event?

        This must fetch all lagging messages from Redis, which means
        this cannot be particularly performant for determining the size of large
        lags. The ideal is therefore to set max_count to a sane minimum, thereby
        minimising the number of messages which need to be pulled back.
        Note however, that this method will never return a value larger than
        max_count + the total pending messages.

        If max_count is None then the full lag will be calculated.
How far behind is the given listener for the given api/event?
How far behind is the given listener for the given api / event?
async def lag(
        self,
        api_name: str,
        event_name: str,
        listener_name: str,
        include_pending: bool = True,
        max_count: Optional[int] = 100,
    ) -> int:
        """How far behind is the given listener for the given api/event?

        This must fetch all lagging messages from Redis, which means
        this cannot be particularly performant for determining the size of large
        lags. The ideal is therefore to set max_count to a sane minimum, thereby
        minimising the number of messages which need to be pulled back.
        Note however, that this method will never return a value larger than
        max_count + the total pending messages.

        If max_count is None then the full lag will be calculated.
        """
        consumer_group = f"{self.service_name}-{listener_name}".encode("utf8")
        stream_name = self._get_stream_names([(api_name, event_name)])[0]

        with await self.connection_manager() as redis:
            try:
                # Get the groups for the given stream
                groups = await redis.xinfo_groups(stream_name)
            except ReplyError as e:
                if "no such key" in str(e):
                    # Steam doesn't exist, so group doesn't exist
                    groups = []
                else:
                    raise

            for group in groups:
                # Find the group we care about
                if group[b"name"] == consumer_group:
                    if max_count is not None:
                        # We should normally receive the last delivered message in addition to any lagging
                        # messages We take that into account here and is tested by test_lag_max_count_set()
                        max_count += 1

                    # This is the part with poor performance. Get all the lagging messages.
                    messages = await redis.xrange(
                        stream=stream_name, start=group[b"last-delivered-id"], count=max_count
                    )

                    # Messages found, exit our for loop
                    break
            else:
                # For loop didn't find any matching consumer groups, return 0
                return 0

        first_message_id = messages[0][0] if messages else None
        total_messages = len(messages)

        if first_message_id == group[b"last-delivered-id"]:
            # The last delivered message is among the results returned, so
            # take that off our count
            lag = total_messages - 1
        elif max_count is not None and total_messages == max_count:
            # The last delivered message has been deleted, and we received the max_count number of results,
            # which is actually one more than we expected because we incremented max_count above.
            # Therefore fix this by subtracting one
            lag = total_messages - 1
        else:
            # The last delivered message has been deleted,
            # so all returned messages are lagging. We retrieved all messages
            # (rather than limiting with max_count) therefore no need to
            # subtract anything
            lag = total_messages

        if include_pending:
            lag += group[b"pending"]

        return lag
async def lag ( self , api_name : str , event_name : str , listener_name : str , include_pending : bool = True , max_count : Optional [ int ] = 100 , ) -> int : consumer_group = f"{self.service_name}-{listener_name}" . encode ( "utf8" ) stream_name = self . _get_stream_names ( [ ( api_name , event_name ) ] ) [ 0 ] with await self . connection_manager ( ) as redis : try : # Get the groups for the given stream groups = await redis . xinfo_groups ( stream_name ) except ReplyError as e : if "no such key" in str ( e ) : # Steam doesn't exist, so group doesn't exist groups = [ ] else : raise for group in groups : # Find the group we care about if group [ b"name" ] == consumer_group : if max_count is not None : # We should normally receive the last delivered message in addition to any lagging # messages We take that into account here and is tested by test_lag_max_count_set() max_count += 1 # This is the part with poor performance. Get all the lagging messages. messages = await redis . xrange ( stream = stream_name , start = group [ b"last-delivered-id" ] , count = max_count ) # Messages found, exit our for loop break else : # For loop didn't find any matching consumer groups, return 0 return 0 first_message_id = messages [ 0 ] [ 0 ] if messages else None total_messages = len ( messages ) if first_message_id == group [ b"last-delivered-id" ] : # The last delivered message is among the results returned, so # take that off our count lag = total_messages - 1 elif max_count is not None and total_messages == max_count : # The last delivered message has been deleted, and we received the max_count number of results, # which is actually one more than we expected because we incremented max_count above. # Therefore fix this by subtracting one lag = total_messages - 1 else : # The last delivered message has been deleted, # so all returned messages are lagging. We retrieved all messages # (rather than limiting with max_count) therefore no need to # subtract anything lag = total_messages if include_pending : lag += group [ b"pending" ] return lag