Channels with Redis not working

[aranvir]

Hi, I struggle a bit with getting channels to run with redis and I don't really understand why.

When I run the code below with MemoryChannelBackend it's all fine. I connect to the SSE endpoint and periodically call the post endpoint and I see messages coming in. When I switch it for the redis backend the SSE endpoint stops working. I monitored redis pubsub with their web interface to sanity check that messages are coming in. I made the following observations:

• If I use channels.publish() the message will appear only once in the redis pubsub.
• If I use channels.wait_published() the message will appear every time I call the endpoint. However, I get an internal error each time stating:

  File "E:\Eigene Dateien\Coding\PythonProjects\litestar-events\.venv\Lib\site-packages\litestar\channels\backends\redis.py", line 127, in publish
    await self._publish_script(keys=list(set(channels)), args=[data])
TypeError: object NoneType can't be used in 'await' expression

• So the redis PubSub object seems to be None and not None at the same time since I the messages DO make it to redis pubsub.
• Likewise, when shutting done the app I get the same NoneType error for the on_shutdown hook

    |   File "E:\Eigene Dateien\Coding\PythonProjects\litestar-events\.venv\Lib\site-packages\litestar\channels\backends\redis.py", line 105, in on_shutdown
    |     await self._pub_sub.reset()
    | TypeError: object NoneType can't be used in 'await' expression

• When using the RedisChannelStreamBackend, I get the error only for the publish call, not in the shutdown hook.
• And regardless, I never receive anything from the subscribed topic.

I suspect there is something fishy with the Python redis client, which is why I added the weird /test-redis endpoint. And indeed: while the pubsub is not None check is successful, when I call the subscribe function afterwards, I get the same NoneType error. Also, while I can ping the Redis instance, I cannot ping the Redis PubSub instance.

I really hope that I havent overlooked something......

I'm running:

• Python 3.11
• Redis 7.2.4 via docker (redis/redis-stack:latest)
• Litestar 2.9.1

from dataclasses import dataclass

from redis import Redis
from litestar import Litestar, get, post
from litestar.channels import ChannelsPlugin
from litestar.channels.backends.memory import MemoryChannelsBackend
from litestar.channels.backends.redis import RedisChannelsPubSubBackend
from litestar.response import ServerSentEvent


@dataclass
class Message:
    content: str


@get("/test-redis")
async def redis_check(channels: ChannelsPlugin) -> None:
    channels._backend: RedisChannelsPubSubBackend
    print(channels._backend._pub_sub)
    print(channels._backend._pub_sub.ping())
    channels._backend._pub_sub.execute_command("PUBLISH hello world")
    if channels._backend._pub_sub is not None:
        await channels._backend._pub_sub.subscribe("example")


@post("/msg/{topic:str}")
async def message(topic: str, data: Message, channels: ChannelsPlugin) -> None:
    channels.publish(data=data.content, channels=[topic])
    # await channels.wait_published(data=data.content, channels=[topic])


@get("/msg/{topic:str}")
async def listener(topic: str, channels: ChannelsPlugin) -> ServerSentEvent:
    async def generator():
        async with channels.start_subscription([topic]) as subscriber:
            async for event in subscriber.iter_events():
                yield event
    return ServerSentEvent(generator())


redis = Redis(host="localhost", port=6379)
print(redis.ping())
backend = RedisChannelsPubSubBackend(redis=redis)


app = Litestar(
    route_handlers=[
        redis_check,
        message,
        listener
    ],
    plugins=[
        ChannelsPlugin(
            backend=backend,
            arbitrary_channels_allowed=True
        )
    ],
    debug=True
)

[provinzkraut]

You're using the sync redis.Redis. You need to use redis.asyncio.Redis.

[aranvir]

🤦‍♂️...thank you... works!