(
        self, streams, consumer_group, expected_events, forever
    )
Coroutine to consume new messages

        The consumption has two stages:

          1. Fetch and yield any messages this consumer is responsible for processing but has yet
             to successfully process. This can happen in cases where a message was
             previously consumed but not acknowledged (i.e. due to an error).
             This is a one-off startup stage.
          2. Wait for new messages to arrive. Yield these messages when they arrive, then
             resume waiting for messages

        See Also:

            _reclaim_lost_messages() - Another coroutine which reclaims messages which timed out
                                       while being processed by other consumers in this group
Coroutine to consume new messages
Coroutine to consume new messages
async def _fetch_new_messages(
        self, streams, consumer_group, expected_events, forever
    ) -> AsyncGenerator[List[EventMessage], None]:
        """Coroutine to consume new messages

        The consumption has two stages:

          1. Fetch and yield any messages this consumer is responsible for processing but has yet
             to successfully process. This can happen in cases where a message was
             previously consumed but not acknowledged (i.e. due to an error).
             This is a one-off startup stage.
          2. Wait for new messages to arrive. Yield these messages when they arrive, then
             resume waiting for messages

        See Also:

            _reclaim_lost_messages() - Another coroutine which reclaims messages which timed out
                                       while being processed by other consumers in this group

        """
        with await self.connection_manager() as redis:
            # Firstly create the consumer group if we need to
            await self._create_consumer_groups(streams, redis, consumer_group)

            # Get any messages that this consumer has yet to process.
            # This can happen in the case where the processes died before acknowledging.
            pending_messages = await redis.xread_group(
                group_name=consumer_group,
                consumer_name=self.consumer_name,
                streams=list(streams.keys()),
                # Using ID '0' indicates we want unacked pending messages
                latest_ids=["0"] * len(streams),
                timeout=None,  # Don't block, return immediately
            )

            event_messages = []
            for stream, message_id, fields in pending_messages:
                message_id = decode(message_id, "utf8")
                stream = decode(stream, "utf8")
                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
                    await redis.xack(stream, consumer_group, message_id)
                    continue

                logger.debug(
                    LBullets(
                        L(
                            "⬅ Receiving pending event {} on stream {}",
                            Bold(message_id),
                            Bold(stream),
                        ),
                        items=dict(
                            **event_message.get_metadata(), kwargs=event_message.get_kwargs()
                        ),
                    )
                )
                event_messages.append(event_message)

            if event_messages:
                yield event_messages

            # We've now cleaned up any old messages that were hanging around.
            # Now we get on to the main loop which blocks and waits for new messages

            while True:
                # Fetch some messages.
                # This will block until there are some messages available
                try:
                    stream_messages = await redis.xread_group(
                        group_name=consumer_group,
                        consumer_name=self.consumer_name,
                        streams=list(streams.keys()),
                        # Using ID '>' indicates we only want new messages which have not
                        # been passed to other consumers in this group
                        latest_ids=[">"] * len(streams),
                        count=self.batch_size,
                    )
                except asyncio.CancelledError:
                    # We need to manually close the connection here otherwise the aioredis
                    # pool will emit warnings saying that this connection still has pending
                    # commands (i.e. the above blocking pop)
                    redis.close()
                    raise

                # Handle the messages we have received
                event_messages = []
                for stream, message_id, fields in stream_messages:
                    message_id = decode(message_id, "utf8")
                    stream = decode(stream, "utf8")
                    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 streaming"
                            " of messages"
                        )
                        await redis.xack(stream, consumer_group, message_id)
                        continue

                    logger.debug(
                        LBullets(
                            L(
                                "⬅ Received new event {} on stream {}",
                                Bold(message_id),
                                Bold(stream),
                            ),
                            items=dict(
                                **event_message.get_metadata(), kwargs=event_message.get_kwargs()
                            ),
                        )
                    )

                    event_messages.append(event_message)

                if event_messages:
                    yield event_messages

                if not forever:
                    return
async def _fetch_new_messages ( self , streams , consumer_group , expected_events , forever ) -> AsyncGenerator [ List [ EventMessage ] , None ] : with await self . connection_manager ( ) as redis : # Firstly create the consumer group if we need to await self . _create_consumer_groups ( streams , redis , consumer_group ) # Get any messages that this consumer has yet to process. # This can happen in the case where the processes died before acknowledging. pending_messages = await redis . xread_group ( group_name = consumer_group , consumer_name = self . consumer_name , streams = list ( streams . keys ( ) ) , # Using ID '0' indicates we want unacked pending messages latest_ids = [ "0" ] * len ( streams ) , timeout = None , # Don't block, return immediately ) event_messages = [ ] for stream , message_id , fields in pending_messages : message_id = decode ( message_id , "utf8" ) stream = decode ( stream , "utf8" ) 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 await redis . xack ( stream , consumer_group , message_id ) continue logger . debug ( LBullets ( L ( "⬅ Receiving pending event {} on stream {}", Bold ( message_id ) , Bold ( stream ) , ) , items = dict ( * * event_message . get_metadata ( ) , kwargs = event_message . get_kwargs ( ) ) , ) ) event_messages . append ( event_message ) if event_messages : yield event_messages # We've now cleaned up any old messages that were hanging around. # Now we get on to the main loop which blocks and waits for new messages while True : # Fetch some messages. # This will block until there are some messages available try : stream_messages = await redis . xread_group ( group_name = consumer_group , consumer_name = self . consumer_name , streams = list ( streams . keys ( ) ) , # Using ID '>' indicates we only want new messages which have not # been passed to other consumers in this group latest_ids = [ ">" ] * len ( streams ) , count = self . batch_size , ) except asyncio . CancelledError : # We need to manually close the connection here otherwise the aioredis # pool will emit warnings saying that this connection still has pending # commands (i.e. the above blocking pop) redis . close ( ) raise # Handle the messages we have received event_messages = [ ] for stream , message_id , fields in stream_messages : message_id = decode ( message_id , "utf8" ) stream = decode ( stream , "utf8" ) 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 streaming" " of messages" ) await redis . xack ( stream , consumer_group , message_id ) continue logger . debug ( LBullets ( L ( "⬅ Received new event {} on stream {}", Bold ( message_id ) , Bold ( stream ) , ) , items = dict ( * * event_message . get_metadata ( ) , kwargs = event_message . get_kwargs ( ) ) , ) ) event_messages . append ( event_message ) if event_messages : yield event_messages if not forever : return