<s> from threading import Thread <EOL> import logging <EOL> from analytics . version import VERSION <EOL> from analytics . request import post <EOL> class Consumer ( Thread ) : <EOL> """<STR_LIT>""" <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> def __init__ ( self , queue , write_key , upload_size = <NUM_LIT:100> , on_error = None ) : <EOL> """<STR_LIT>""" <EOL> Thread . __init__ ( self ) <EOL> self . daemon = True <EOL> self . upload_size = upload_size <EOL> self . write_key = write_key <EOL> self . on_error = on_error <EOL> self . queue = queue <EOL> self . retries = <NUM_LIT:3> <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . log . debug ( '<STR_LIT>' ) <EOL> self . running = True <EOL> while self . running : <EOL> self . upload ( ) <EOL> self . log . debug ( '<STR_LIT>' ) <EOL> def pause ( self ) : <EOL> """<STR_LIT>""" <EOL> self . running = False <EOL> def upload ( self ) : <EOL> """<STR_LIT>""" <EOL> success = False <EOL> batch = self . next ( ) <EOL> if len ( batch ) == <NUM_LIT:0> : <EOL> return False <EOL> try : <EOL> self . request ( batch ) <EOL> success = True <EOL> except Exception as e : <EOL> self . log . error ( '<STR_LIT>' , e ) <EOL> success = False <EOL> if self . on_error : <EOL> self . on_error ( e , batch ) <EOL> finally : <EOL> for item in batch : <EOL> self . queue . task_done ( ) <EOL> return success <EOL> def next ( self ) : <EOL> """<STR_LIT>""" <EOL> queue = self . queue <EOL> items = [ ] <EOL> item = self . next_item ( ) <EOL> if item is None : <EOL> return items <EOL> items . append ( item ) <EOL> while len ( items ) < self . upload_size and not queue . empty ( ) : <EOL> item = self . next_item ( ) <EOL> if item : <EOL> items . append ( item ) <EOL> return items <EOL> def next_item ( self ) : <EOL> """<STR_LIT>""" <EOL> queue = self . queue <EOL> try : <EOL> item = queue . get ( block = True , timeout = <NUM_LIT:5> ) <EOL> return item <EOL> except Exception : <EOL> return None <EOL> def request ( self , batch , attempt = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> post ( self . write_key , batch = batch ) <EOL> except : <EOL> if attempt > self . retries : <EOL> raise <EOL> self . request ( batch , attempt + <NUM_LIT:1> ) </s>
