<s> import os <EOL> import errno <EOL> from . import constants as _c , FSQWorkItem , path as fsq_path , FSQScanError , FSQCannotLockError , FSQWorkItemError , FSQDownError , FSQError , is_down , hosts as fsq_hosts , host_is_down <EOL> from . internal import wrap_io_os_err <EOL> class FSQScanGenerator ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , queue , item_ids , lock = None , ttl = None , <EOL> max_tries = None , ignore_down = False , no_open = False , <EOL> host = False ) : <EOL> '''<STR_LIT>''' <EOL> self . _index = - <NUM_LIT:1> <EOL> self . item = None <EOL> self . queue = queue <EOL> self . host = host <EOL> self . item_ids = item_ids <EOL> self . lock = _c . FSQ_LOCK if lock is None else lock <EOL> self . ttl = _c . FSQ_TTL if ttl is None else ttl <EOL> self . max_tries = _c . FSQ_MAX_TRIES if max_tries is None else max_tries <EOL> self . ignore_down = ignore_down <EOL> self . no_open = no_open <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def __del__ ( self ) : <EOL> '''<STR_LIT>''' <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> del self . item <EOL> def next ( self ) : <EOL> while self . _index < len ( self . item_ids ) - <NUM_LIT:1> : <EOL> self . _index += <NUM_LIT:1> <EOL> if getattr ( self , '<STR_LIT>' , None ) is not None : <EOL> del self . item <EOL> if self . host : <EOL> host = self . item_ids [ self . _index ] [ <NUM_LIT:0> ] <EOL> item = self . item_ids [ self . _index ] [ <NUM_LIT:1> ] <EOL> else : <EOL> host = None <EOL> item = self . item_ids [ self . _index ] <EOL> if not self . ignore_down and is_down ( self . queue ) and ( not host or <EOL> host_is_down ( self . queue , host ) ) : <EOL> raise FSQDownError ( errno . EAGAIN , u'<STR_LIT>' u'<STR_LIT>' . format ( self . queue ) ) <EOL> try : <EOL> self . item = FSQWorkItem ( self . queue , <EOL> item , <EOL> lock = self . lock , ttl = self . ttl , <EOL> max_tries = self . max_tries , <EOL> no_open = self . no_open , <EOL> host = host ) <EOL> except ( FSQWorkItemError , FSQCannotLockError , ) , e : <EOL> if e . errno == errno . EAGAIN or e . errno == errno . ENOENT : <EOL> continue <EOL> raise e <EOL> return self . item <EOL> if getattr ( self , '<STR_LIT>' , None ) is not None : <EOL> del self . item <EOL> raise StopIteration ( ) <EOL> def scan_forever ( queue , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> process_once_now = kwargs . get ( '<STR_LIT>' , True ) <EOL> if process_once_now : <EOL> for work in scan ( queue , * args , ** kwargs ) : <EOL> yield work <EOL> while True : <EOL> with open ( fsq_path . trigger ( queue ) , '<STR_LIT:rb>' ) as t : <EOL> t . read ( <NUM_LIT:1> ) <EOL> for work in scan ( queue , * args , ** kwargs ) : <EOL> yield work <EOL> def scan ( queue , lock = None , ttl = None , max_tries = None , ignore_down = False , <EOL> no_open = False , generator = FSQScanGenerator , host = False , hosts = None ) : <EOL> '''<STR_LIT>''' <EOL> lock = _c . FSQ_LOCK if lock is None else lock <EOL> ttl = _c . FSQ_TTL if lock is None else ttl <EOL> max_tries = _c . FSQ_MAX_TRIES if max_tries is None else max_tries <EOL> item_ids = [ ] <EOL> try : <EOL> if not host and hosts is None : <EOL> item_ids = os . listdir ( fsq_path . queue ( queue ) ) <EOL> item_ids . sort ( ) <EOL> else : <EOL> if hosts is None : <EOL> hosts = fsq_hosts ( queue ) <EOL> for trg_host in hosts : <EOL> for item in os . listdir ( fsq_path . queue ( queue , trg_host ) ) : <EOL> item_ids . append ( ( trg_host , item ) ) <EOL> item_ids . sort ( key = lambda x : x [ <NUM_LIT:1> ] ) <EOL> except ( OSError , IOError , ) , e : <EOL> if e . errno == errno . ENOENT : <EOL> raise FSQScanError ( e . errno , u'<STR_LIT>' u'<STR_LIT>' . format ( queue ) ) <EOL> elif isinstance ( e , FSQError ) : <EOL> raise e <EOL> raise FSQScanError ( e . errno , wrap_io_os_err ( e ) ) <EOL> return generator ( queue , item_ids , lock = lock , ttl = ttl , max_tries = max_tries , <EOL> no_open = no_open , host = host ) </s>
