|
def get_next_message(self): |
|
projects = list(self.redis.smembers(self.projects_key)) |
|
if not projects: |
|
return None |
|
|
|
project_id = random.choice(projects) |
|
msg = self.redis.rpop(self.get_queue_key(project_id)) |
|
if not msg: |
|
return None |
get_next_message() uses redis.smembers() to get a list of all project_id's, and project_ids are added to the set in put(), but project_id's are never removed from the set if there are no request in the project_id's queue. So get_next_message() may be calling redis.rpop() on a key that no longer exists.
Another option may be to use redis.scan() for keys that match "cyclops:queue:*", and randomly select from that list for the next message. This would guarantee that redis.rpop() is called on a queue that contains at least one message.
cyclops/cyclops/storage.py
Lines 95 to 103 in 9136a4f
get_next_message() uses redis.smembers() to get a list of all project_id's, and project_ids are added to the set in put(), but project_id's are never removed from the set if there are no request in the project_id's queue. So get_next_message() may be calling redis.rpop() on a key that no longer exists.
Another option may be to use redis.scan() for keys that match "cyclops:queue:*", and randomly select from that list for the next message. This would guarantee that redis.rpop() is called on a queue that contains at least one message.