#NEWLINE# def run( self , file , request , threadpool , chunk_size = constants . DEFAULT_CHUNK_SIZE , send_ahead = constants . SEND_AHEAD ) : #NEWLINE# #INDENT# self . watch ( request ) #NEWLINE# seek_ptr = 0 #NEWLINE# chunks_sent = 0 #NEWLINE# other_received = 0 #NEWLINE# with open ( file , 'rb' ) as f : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# chunk = read_file_async ( threadpool , f , limit = chunk_size ) #NEWLINE# more_coming = ( len ( chunk ) > 0 ) #NEWLINE# ( request << ( 'chunk' , chunk , more_coming , f ) ) #NEWLINE# seek_ptr += len ( chunk ) #NEWLINE# chunks_sent += 1 #NEWLINE# if ( not more_coming ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# try : #NEWLINE# #INDENT# timeout = ( 0 if ( ( seek_ptr - other_received ) < send_ahead ) else None ) #NEWLINE# msg = self . get ( OR ( ( 'terminated' , request ) , ( 'received' , ANY ) ) , timeout = timeout ) #NEWLINE# #UNINDENT# except Empty : #NEWLINE# #INDENT# continue #NEWLINE# #UNINDENT# if ( msg == ( 'received' , ANY ) ) : #NEWLINE# #INDENT# ( _ , other_received ) = msg #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# while True : #NEWLINE# #INDENT# msg = self . get ( OR ( ( 'received' , ANY ) , ( 'terminated' , request ) ) , timeout = 10.0 ) #NEWLINE# if ( msg [ 0 ] == 'terminated' ) : #NEWLINE# #INDENT# break