<s> from . base import BaseWorker , BaseManager <EOL> from . . scheduler . gevent_scheduler import Scheduler <EOL> from . . import settings <EOL> import gevent <EOL> from gevent import Greenlet <EOL> from gevent . event import Event <EOL> from gevent . queue import Queue <EOL> from Queue import Empty <EOL> import logging <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> from django . utils . timezone import now <EOL> class Worker ( BaseWorker ) : <EOL> def _name ( self ) : <EOL> return self . greenlet_name <EOL> def set_name ( self , name ) : <EOL> self . greenlet_name = name <EOL> class ACKWatcher ( Greenlet ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , queue , stop_event , callback ) : <EOL> super ( ACKWatcher , self ) . __init__ ( ) <EOL> self . queue = queue <EOL> self . stop_event = stop_event <EOL> self . callback = callback <EOL> def _run ( self ) : <EOL> while not self . stop_event . is_set ( ) : <EOL> try : <EOL> ack_uuid = self . queue . get ( True , <NUM_LIT:1> ) <EOL> self . callback ( ack_uuid ) <EOL> except Empty : <EOL> pass <EOL> class GreenletManager ( BaseManager ) : <EOL> greenlet = True <EOL> process_list = [ ] <EOL> def __init__ ( self ) : <EOL> super ( GreenletManager , self ) . __init__ ( ) <EOL> self . stop_event = Event ( ) <EOL> self . work_queue = Queue ( settings . GREENQUEUE_BACKEND_TASKBUFF ) <EOL> self . ack_queue = Queue ( ) <EOL> def start_greenlet_pool ( self ) : <EOL> for i in xrange ( settings . GREENQUEUE_BACKEND_POOLSIZE ) : <EOL> log . info ( "<STR_LIT>" . format ( i ) ) <EOL> worker = Worker ( self . work_queue , self . ack_queue , self . stop_event ) <EOL> worker . set_name ( "<STR_LIT>" . format ( i ) ) <EOL> g = Greenlet . spawn ( worker ) <EOL> self . process_list . append ( g ) <EOL> def start ( self ) : <EOL> self . watcher = ACKWatcher ( self . ack_queue , self . stop_event , self . _ack_callback ) <EOL> self . watcher . start ( ) <EOL> self . scheduler = Scheduler ( self . stop_event , self . _handle_message ) <EOL> self . scheduler . start ( ) <EOL> self . start_greenlet_pool ( ) <EOL> def _handle_message ( self , name , uuid , args , kwargs , message = { } ) : <EOL> now_date = now ( ) <EOL> if "<STR_LIT>" in message : <EOL> eta = self . parse_eta ( message [ '<STR_LIT>' ] ) <EOL> if now ( ) < eta : <EOL> log . debug ( "<STR_LIT>" , eta . isoformat ( ) ) <EOL> self . scheduler . push_task ( eta , ( name , uuid , args , kwargs ) ) <EOL> return <EOL> self . work_queue . put ( ( name , uuid , args , kwargs ) , block = True ) <EOL> def close ( self ) : <EOL> self . stop_event . set ( ) </s>
