#NEWLINE# def _clean( self ) : #NEWLINE# #INDENT# if ( not self . _ongoing_requests ) : #NEWLINE# #INDENT# return #NEWLINE# #UNINDENT# with self . _ongoing_requests_lock : #NEWLINE# #INDENT# ongoing_requests_uuids = set ( six . iterkeys ( self . _ongoing_requests ) ) #NEWLINE# #UNINDENT# waiting_requests = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# expired_requests = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } #NEWLINE# for request_uuid in ongoing_requests_uuids : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# request = self . _ongoing_requests [ request_uuid ] #NEWLINE# #UNINDENT# except KeyError : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if request . expired : #NEWLINE# #INDENT# expired_requests [ request_uuid ] = request #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( request . current_state == pr . WAITING ) : #NEWLINE# #INDENT# waiting_requests [ request_uuid ] = request #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# if expired_requests : #NEWLINE# #INDENT# with self . _ongoing_requests_lock : #NEWLINE# #INDENT# while expired_requests : #NEWLINE# #INDENT# ( request_uuid , request ) = expired_requests . popitem ( ) #NEWLINE# if self . _handle_expired_request ( request ) : #NEWLINE# #INDENT# del self . _ongoing_requests [ request_uuid ] #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# #UNINDENT# if waiting_requests : #NEWLINE# #INDENT# finder = self . _finder #NEWLINE# new_messages_processed = finder . messages_processed #NEWLINE# last_messages_processed = self . _messages_processed [ 'finder' ] #NEWLINE# if ( new_messages_processed > last_messages_processed ) : #NEWLINE# #INDENT# while waiting_requests : #NEWLINE# #INDENT# ( _request_uuid , request ) = waiting_requests . popitem ( ) #NEWLINE# worker = finder . get_worker_for_task ( request . task ) #NEWLINE# if ( ( worker is not None ) and request . transition_and_log_error ( pr . PENDING , logger = LOG ) ) : #NEWLINE# #INDENT# self . _publish_request ( request , worker ) #NEWLINE# #UNINDENT# #UNINDENT# self . _messages_processed [ 'finder' ] = new_messages_processed