sjhewitt/celery_redis

celery_redis's Issues

timeout

Here is a version that works well for me. it uses the new timeout option on get_message so it's
timely and no polling. on redis_py 2.10.3, get_message return twice, once very quickly and once at the message or at the timeout. either way, it's in a loop so it works:

def wait_for(self, task_id,
                 timeout=None, interval=0.5, no_ack=True, propagate=True,on_interval=None):
        key = self.get_key_for_task(task_id)
        p = self.client.pubsub(ignore_subscribe_messages=True)
        p.subscribe(key)
        meta = self.client.get(key)
        # check the result, if not there, start listening on pubsub queue
        if meta:
            meta = self.decode(meta)
            if meta['status'] not in states.READY_STATES:
                meta = None
        if not meta:
            if timeout:
                time_elapsed = 0.0
                while True:
                    start = time.time()
                    message = p.get_message(timeout=timeout)
                    if message:
                        break
                    if on_interval:
                        on_interval()
                    #time.sleep(interval)   <<< NOT NEED, 
                    time_elapsed += interval
                    if time_elapsed >= timeout:
                        raise TimeoutError('The operation timed out.')
            else:
                message = next(p.listen())
            assert message['type'] == 'message'
            assert message['channel'] == key
            meta = self.decode(message['data'])
        p.unsubscribe()
        p.close()
        return meta