subscribing to an empty set kills pubsub

[dfee]

This appears to be a bug, but I'll leave that to the public to decide.

When subscribing to nothing, the pubsub stops listening completely.

subscriber.subscribe([])
subscriber.subscribe(['my_channel'])

After the first statement, subscriber.next_published() will never hear another reply.

This doesn't appear to be documented on Redis Subscribe so I'm assuming this is a bug?

[dfee]

To add a bit more color, these appear to be the situations in which a subscriber stops listening:

1. subscriber.subscribe([])
2. subscriber.unsubscribe([])
3. anytime a subscriber has unsubscribed from its last channel. e.g.

subscriber.subscribe(['my_channel'])
subscriber.unsubscribe(['my_channel'])
subscriber.subscribe(['my_channel'])  # works, but we'll never hear a message.

I've validated with redis-cli that the messages are getting pushed to client (via the return value of PUBLISH), so I know that subscriber.next_published() should be getting something.

[dfee]

Here's an example that will demonstrate this nicely:

import asyncio
import asyncio_redis


CHANNEL = 'my_channel'

async def listen(sub, sub_name):
    for i in range(10):
        print('{}: {}'.format(sub_name, await sub.next_published()))

async def publish_loop(pub_c):
    for i in range(10):
        await pub_c.publish(CHANNEL, 'message {}'.format(i))

async def build_pubsub(host='redis', port=6379):
    sub_c1 = await asyncio_redis.connection.Connection.create(host, port)
    sub_c2 = await asyncio_redis.connection.Connection.create(host, port)
    sub1 = await sub_c1.start_subscribe()
    sub2 = await sub_c2.start_subscribe()
    pub_c = await asyncio_redis.connection.Connection.create(host, port)
    return pub_c, sub1, sub2

async def test_pubsub():
    pub_c, sub_good, sub_bad = await build_pubsub()

    # Subscribe to channels
    await sub_good.subscribe([CHANNEL])
    await sub_bad.subscribe([])
    await sub_bad.subscribe([CHANNEL])

    # Start listeners
    asyncio.ensure_future(listen(sub_good, 'sub_good'))
    asyncio.ensure_future(listen(sub_bad, 'sub_bad'))

    # Publish messages
    await publish_loop(pub_c)

loop = asyncio.get_event_loop()
loop.run_until_complete(test_pubsub())

Results in:

sub_good: PubSubReply(channel='my_channel', value='message 0')
sub_good: PubSubReply(channel='my_channel', value='message 1')
sub_good: PubSubReply(channel='my_channel', value='message 2')
sub_good: PubSubReply(channel='my_channel', value='message 3')
sub_good: PubSubReply(channel='my_channel', value='message 4')
sub_good: PubSubReply(channel='my_channel', value='message 5')
sub_good: PubSubReply(channel='my_channel', value='message 6')
sub_good: PubSubReply(channel='my_channel', value='message 7')
sub_good: PubSubReply(channel='my_channel', value='message 8')
sub_good: PubSubReply(channel='my_channel', value='message 9')