The fix has a bug (subscribe happens too frequently)
super().__init__(**kwargs)
        self.channels = channels
        self.redis_conn_id = redis_conn_id
    @cached_property
    def pubsub(self):
        return RedisHook(redis_conn_id=self.redis_conn_id).get_conn().pubsub()
    def poke(self, context: Context) -> bool:
        """
        Check for message on subscribed channels and write to xcom the message with key ``message``.
        An example of message ``{'type': 'message', 'pattern': None, 'channel': b'test', 'data': b'hello'}``
        :param context: the context object
        :return: ``True`` if message (with type 'message') is available or ``False`` if not
        """
        self.log.info("RedisPubSubSensor checking for message on channels: %s", self.channels)
        self.pubsub.subscribe(self.channels)
        message = self.pubsub.get_message()
        self.log.info("Message %s from channel %s", message, self.channels)
        
        if message and message["type"] == "message":
            context["ti"].xcom_push(key="message", value=message)
            self.pubsub.unsubscribe(self.channels)
            return True
        return False