(
        self,
        fields: dict,
        expected_event_names: Iterable[str],
        stream: str,
        native_id: str,
        consumer_group: Optional[str],
    )
return message
Convert a dict of Redis message fields into a RedisEventMessage
Convert a dict of Redis message fields into a RedisEventMessage
Convert a dict of Redis message fields into a RedisEventMessage
def _fields_to_message(
        self,
        fields: dict,
        expected_event_names: Iterable[str],
        stream: str,
        native_id: str,
        consumer_group: Optional[str],
    ) -> Optional[RedisEventMessage]:
        """Convert a dict of Redis message fields into a RedisEventMessage"""

        if not fields:
            # Redis returned us a message of 'nil'
            raise NoopMessage()

        if tuple(fields.items()) == ((b"", b""),):
            # Is a noop message, ignore
            raise NoopMessage()

        message = self.deserializer(
            fields, stream=stream, native_id=native_id, consumer_group=consumer_group
        )

        want_message = ("*" in expected_event_names) or (message.event_name in expected_event_names)
        if self.stream_use == StreamUse.PER_API and not want_message:
            # Only care about events we are listening for. If we have one stream
            # per API then we're probably going to receive some events we don't care about.
            logger.debug(
                f"Ignoring message for unneeded event: {message}. "
                f"Only listening for {', '.join(expected_event_names)}"
            )
            raise IgnoreMessage()
        return message
def _fields_to_message ( self , fields : dict , expected_event_names : Iterable [ str ] , stream : str , native_id : str , consumer_group : Optional [ str ] , ) -> Optional [ RedisEventMessage ] : if not fields : # Redis returned us a message of 'nil' raise NoopMessage ( ) if tuple ( fields . items ( ) ) == ( ( b"" , b"" ) , ) : # Is a noop message, ignore raise NoopMessage ( ) message = self . deserializer ( fields , stream = stream , native_id = native_id , consumer_group = consumer_group ) want_message = ( "*" in expected_event_names ) or ( message . event_name in expected_event_names ) if self . stream_use == StreamUse . PER_API and not want_message : # Only care about events we are listening for. If we have one stream # per API then we're probably going to receive some events we don't care about. logger . debug ( f"Ignoring message for unneeded event: {message}. " f"Only listening for {', '.join(expected_event_names)}" ) raise IgnoreMessage ( ) return message