[2.0] xread() halts when the connection is lost.

[marciorasf]

Describe the bug

If the connection with Redis is lost while the program is awaiting for xread() it get stuck until the connection is reestablished.
It doesn't raise any Exception, or at least timeout after the block time.

If I wrap xread() with asyncio.wait_for(), and force a disconnection, the wait_for() timeouts correctly, and the subsequent xread() calls raise ConnectionError. So it confirms that the bug only happens if the code is awaiting xread() when the disconnection occurs.

To Reproduce

1. You need a running Redis with a stream with entries
2. Set correct values for redis_dsn and stream. Then run:

import asyncio
import logging
from datetime import datetime

import aioredis

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

redis_dsn = "redis://localhost:6379"
stream = "stream_name"


async def main() -> None:
    redis = aioredis.from_url(redis_dsn, decode_responses=True)
    index = "0"

    while True:
        try:
            response = await redis.xread(
                {stream: index},
                block=1000,
                count=10,
            )

            logger.info(
                f"{datetime.now()} : Read {len(response)} messages from stream."
            )

            for _stream_name, messages in response:
                for message_id, values in messages:
                    index = message_id
        except Exception as err:
            logger.warn(err)


asyncio.run(main())

3. The logs should be displayed as the entries are read.
4. Force a disconnection, such as stopping the Redis.
5. You should see that the logs have stopped and no exception is logged.
6. [Optional] If you force a reconnection, the program should resume so it start to display the logs again.

Expected behavior

After running the snippet, you should see the logs been displayed rapidly. After the disconnection the program will halt, so the logs will also stop, and no exception is raised.

Logs/tracebacks

It's not applicable.

Python Version

Python 3.8.10

aioredis Version

aioredis 2.0.0

Additional context

No response

Code of Conduct

• I agree to follow the aio-libs Code of Conduct

[Andrew-Chen-Wang]

Hopefully this is resolved by #1156 as this seems to relate to #778