(self, stream_names: List[str])
Cleanup old consumers and groups

        A group will be deleted if it contains no consumers.

        A consumer will be deleted if it has been idle for more than consumer_ttl.
Cleanup old consumers and groups
Cleanup old consumers and groups
async def _cleanup(self, stream_names: List[str]):
        """Cleanup old consumers and groups

        A group will be deleted if it contains no consumers.

        A consumer will be deleted if it has been idle for more than consumer_ttl.
        """
        if not self.consumer_ttl:
            # Don't do the cleanup if no TTL is given, consider this to mean
            # cleanup is disabled
            return

        with await self.connection_manager() as redis:
            # For every stream key...
            for stream_name in stream_names:
                consumers: List[Tuple[str, str]] = []

                # Get all the groups for that key...
                try:
                    groups = await redis.xinfo_groups(stream_name)
                except ReplyError as e:
                    if "ERR no such key" in str(e):
                        # Steam doesn't exist yet
                        groups = []
                    else:
                        raise

                for group in groups:
                    active_consumers = 0
                    group_name = group[b"name"]

                    try:
                        consumers = await redis.xinfo_consumers(stream_name, group_name)
                    except ReplyError as e:
                        # Group has already vanished,
                        # perhaps something else is doing a cleanup
                        if "NOGROUP" in str(e):
                            continue

                    # Get all the consumers for that group
                    for consumer in consumers:
                        consumer_name = consumer[b"name"]
                        idle_seconds = consumer[b"idle"] / 1000

                        # And delete the consumer if they have not re-started
                        # listening for self.consumer_ttl seconds
                        if idle_seconds >= self.consumer_ttl:
                            logger.debug(
                                f"Cleaning up consumer {consumer_name} in group {group_name} on"
                                f" stream {stream_name}. The consumer has been idle for"
                                f" {idle_seconds} seconds, which is more than the consumer TTL of"
                                f" {self.consumer_ttl}"
                            )
                            await redis.xgroup_delconsumer(stream_name, group_name, consumer_name)
                        else:
                            active_consumers += 1

                    # If no active consumers were found for this group, then delete the entire group
                    # on the grounds that it is no longer used and can be cleaned up.
                    if not active_consumers:
                        # We do this atomically using a lua script. This avoids race conditions
                        # whereby a new consumer comes into existence the moment before we delete the group
                        try:
                            await redis.eval(
                                ATOMIC_DESTROY_CONSUMER_GROUP, [stream_name], [group_name]
                            )
                        except ReplyError as e:
                            if "NOGROUP" in str(e):
                                # Already deleted
                                pass
async def _cleanup ( self , stream_names : List [ str ] ) : if not self . consumer_ttl : # Don't do the cleanup if no TTL is given, consider this to mean # cleanup is disabled return with await self . connection_manager ( ) as redis : # For every stream key... for stream_name in stream_names : consumers : List [ Tuple [ str , str ] ] = [ ] # Get all the groups for that key... try : groups = await redis . xinfo_groups ( stream_name ) except ReplyError as e : if "ERR no such key" in str ( e ) : # Steam doesn't exist yet groups = [ ] else : raise for group in groups : active_consumers = 0 group_name = group [ b"name" ] try : consumers = await redis . xinfo_consumers ( stream_name , group_name ) except ReplyError as e : # Group has already vanished, # perhaps something else is doing a cleanup if "NOGROUP" in str ( e ) : continue # Get all the consumers for that group for consumer in consumers : consumer_name = consumer [ b"name" ] idle_seconds = consumer [ b"idle" ] / 1000 # And delete the consumer if they have not re-started # listening for self.consumer_ttl seconds if idle_seconds >= self . consumer_ttl : logger . debug ( f"Cleaning up consumer {consumer_name} in group {group_name} on" f" stream {stream_name}. The consumer has been idle for" f" {idle_seconds} seconds, which is more than the consumer TTL of" f" {self.consumer_ttl}" ) await redis . xgroup_delconsumer ( stream_name , group_name , consumer_name ) else : active_consumers += 1 # If no active consumers were found for this group, then delete the entire group # on the grounds that it is no longer used and can be cleaned up. if not active_consumers : # We do this atomically using a lua script. This avoids race conditions # whereby a new consumer comes into existence the moment before we delete the group try : await redis . eval ( ATOMIC_DESTROY_CONSUMER_GROUP , [ stream_name ] , [ group_name ] ) except ReplyError as e : if "NOGROUP" in str ( e ) : # Already deleted pass