The fix has a bug (subscribe happens too frequently)
#
#
#
from __future__ import annotations
from functools import cached_property
from typing import TYPE_CHECKING, Sequence
from airflow.providers.redis.hooks.redis import RedisHook
from airflow.sensors.base import BaseSensorOperator
if TYPE_CHECKING:
    from airflow.utils.context import Context
class RedisPubSubSensor(BaseSensorOperator):
    """
    Redis sensor for reading a message from pub sub channels.
    :param channels: The channels to be subscribed to (templated)
    :param redis_conn_id: the redis connection id
    """
    template_fields: Sequence[str] = ("channels",)
    ui_color = "#f0eee4"
    def __init__(self, *, channels: list[str] | str, redis_conn_id: str, **kwargs) -> None: