(self, streams, redis, consumer_group)
Ensure the consumer groups exist

        This is means we have to ensure the streams exist too
Ensure the consumer groups exist
Ensure the consumer groups exist
async def _create_consumer_groups(self, streams, redis, consumer_group):
        """Ensure the consumer groups exist

        This is means we have to ensure the streams exist too
        """
        for stream, since in streams.items():
            if not await redis.exists(stream):
                # Add a noop to ensure the stream exists
                # TODO: We can now use MKSTREAM, change this logic
                #       Documented here: https://redis.io/topics/streams-intro
                await redis.xadd(stream, fields={"": ""})

            try:
                # Create the group (it may already exist)
                await redis.xgroup_create(stream, consumer_group, latest_id=since)
            except ReplyError as e:
                if "BUSYGROUP" in str(e):
                    # Already exists
                    pass
                else:
                    raise
async def _create_consumer_groups ( self , streams , redis , consumer_group ) : for stream , since in streams . items ( ) : if not await redis . exists ( stream ) : # Add a noop to ensure the stream exists # TODO: We can now use MKSTREAM, change this logic #       Documented here: https://redis.io/topics/streams-intro await redis . xadd ( stream , fields = { "" : "" } ) try : # Create the group (it may already exist) await redis . xgroup_create ( stream , consumer_group , latest_id = since ) except ReplyError as e : if "BUSYGROUP" in str ( e ) : # Already exists pass else : raise