#NEWLINE# def run( self ) : #NEWLINE# #INDENT# my_name = threading . current_thread ( ) . name #NEWLINE# print ( '+ KinesisWorker:' , my_name ) #NEWLINE# print ( '+-> working with iterator:' , self . iterator_type ) #NEWLINE# response = kinesis . get_shard_iterator ( self . stream_name , self . shard_id , self . iterator_type ) #NEWLINE# next_iterator = response [ 'ShardIterator' ] #NEWLINE# print ( '+-> getting next records using iterator:' , next_iterator ) #NEWLINE# start = datetime . datetime . now ( ) #NEWLINE# finish = ( start + datetime . timedelta ( seconds = self . worker_time ) ) #NEWLINE# while ( finish > datetime . datetime . now ( ) ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# response = kinesis . get_records ( next_iterator , limit = 25 ) #NEWLINE# self . total_records += len ( response [ 'Records' ] ) #NEWLINE# if ( len ( ptee [ 'Records' ] ) > 0 ) : #NEWLINE# #INDENT# print ( '\n+-> {1} Got {0} Worker Records' . format ( len ( response [ 'Records' ] ) , my_name ) ) #NEWLINE# if self . echo : #NEWLINE# #INDENT# echo_records ( response [ 'Records' ] ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# find_eggs ( response [ 'Records' ] ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# sys . stdout . write ( '.' ) #NEWLINE# sys . stdout . flush ( ) #NEWLINE# #UNINDENT# next_iterator = response [ 'NextShardIterator' ] #NEWLINE# time . sleep ( self . sleep_interval ) #NEWLINE# #UNINDENT# except ProvisionedThroughputExceededException as ptee : #NEWLINE# #INDENT# print ( ptee . message ) #NEWLINE# time . sleep ( 5 )