Not all published messages are processed

[mattlong]

If I publish multiple messages in quick succession in a custom Redis Store's publish_message method, only the first message is received by the browser. I've traced the problem to select() not being triggered for the additional messages due to redis-py using buffered reads within it's PubSub parse_response method. redis-py calls readline on it's socket which will consume multiple messages off its socket but only return the first while leaving any additional messages buffered. Since the subsequent messages have already been read and buffered, select() is (correctly) not being triggered again.

[jrief]

Does this mean, that in your use case more than one message is retrievable from

line 81: response = redis_store.parse_response() 

Hmmm, yes this could be possible. I will have a deeper look.

[ttrifonov]

I also hit this but with having multiple subscriptions. parse_response() reads only first message, and if there are others - they are not read, nor select returns the socket, exactly as described by mattlong. I've made a quick workaround with using non-blocking connection and doing multiple reads with catching the socket exception, but as long as redis-py is not supposed to work with non-blocking connections, this is just an ugly hack yet..

[mattlong]

@jrief Yes, exactly: more than one message is retrievable from redis_store.parse_response

@ttrifonov My current work around is to monkey patch redis-py's redis.connectin.PythonParser's read method to do unbuffered reading instead of using readline. Could you share what your workaround looks like? Here's mine:

# no length, read a full line
# readline is buffered, so #badnewsbears
#return self._fp.readline()[:-2]

# let's do it totally unbuffered instead!
buf = BytesIO()
byte = None
while byte != '\n':
    byte = self._fp.read(1)
    buf.write(byte)
buf.seek(0)
return buf.read()[:-2]

[ttrifonov]

I've did a quick patch from blocking to non-blocking connection with patching the Connection class through ConnectionPool(connection_class=FlexiBlockingConnection)

class FlexiBlockingConnection(Connection):

def _connect(self):
    sock = super(FlexiBlockingConnection, self)._connect()
    # this line here is the difference
    sock.settimeout(0.1)
    return sock

def read_response(self):
    "Read the response from a previously sent command"
    try:
        response = self._parser.read_response()
    except ConnectionError, cex:
        return None
    except:
        self.disconnect()
        raise
    if isinstance(response, ResponseError):
        raise response
    return response

and for reading messages I have:

while True:
    response = self.sub.connection.read_response()
    if not response:
        break
    yield response

which will stop consuming the response on Connection error and since the data is already buffered, the non-blocking reads work fine. A really ugly patch, but I had no time to investigate in depth how to better patch the redis-py implementation.

As as side effect, I also have to retry few times the subscribe and unsubscribe commands sometimes.

I'll try your patch, looks much better.

[ttrifonov]

On second thought, this issue might be probably addressed to Redis-py developers instead?

[mattlong]

A simple repro for this is to start with the chatserver example and subclass RedisStore such that publish_message publishes multiple messages back-to-back to the Redis pubsub connection:

from ws4redis.store import RedisStore

class SuchRedisStore(RedisStore):
    # rest of class definition...

    def publish_message(self, message):
        num_to_send = int(message)
        for n in range(1, num_to_send + 1):
            message = 'Message %s of %s' % (n, num_to_send)
            self._connection.publish('_broadcast_:foobar', message)

Then send an int greater than one from the chatroom and observe that not all messages are shown back to you that should be.

[jrief]

Just a notice:
The latest version of django-websocket-redis has a build-in publisher class, so there is not more need to use the raw Redis connector inside user space code.

But anyway, thanks, I will have a look at this example.

[jrief]

@mattlong
I just reviewed your code. You are reusing the storage bucket with the same identical key. Therefore, from my point of view, what you receive is somehow arbitrary. What would you expect?