#NEWLINE# @ defer . inlineCallbacks #NEWLINE# def wait_for_events( self , user_id , timeout , callback , room_ids = None , from_token = StreamToken . START ) : #NEWLINE# #INDENT# 'Wait until the callback returns a non empty response or the\n        timeout fires.\n        ' #NEWLINE# user_stream = self . user_to_user_stream . get ( user_id ) #NEWLINE# if ( user_stream is None ) : #NEWLINE# #INDENT# appservice = ( yield self . store . get_app_service_by_user_id ( user_id ) ) #NEWLINE# current_token = ( yield self . event_sources . get_current_token ( ) ) #NEWLINE# if ( room_ids is None ) : #NEWLINE# #INDENT# rooms = ( yield self . store . get_rooms_for_user ( user_id ) ) #NEWLINE# room_ids = [ room . room_id for room in rooms ] #NEWLINE# #UNINDENT# user_stream = _NotifierUserStream ( user_id = user_id , rooms = room_ids , appservice = appservice , current_token = current_token , time_now_ms = self . clock . time_msec ( ) ) #NEWLINE# self . _register_with_keys ( user_stream ) #NEWLINE# #UNINDENT# result = None #NEWLINE# if timeout : #NEWLINE# #INDENT# listener = None #NEWLINE# def timed_out( ) : #NEWLINE# #INDENT# if listener : #NEWLINE# #INDENT# listener . deferred . cancel ( ) #NEWLINE# #UNINDENT# #UNINDENT# timer = self . clock . call_later ( ( timeout / 1000.0 ) , timed_out ) #NEWLINE# prev_token = from_token #NEWLINE# while ( not result ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# current_token = user_stream . current_token #NEWLINE# result = ( yield callback ( prev_token , current_token ) ) #NEWLINE# if result : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# prev_token = current_token #NEWLINE# listener = user_stream . new_listener ( prev_token ) #NEWLINE# with PreserveLoggingContext ( ) : #NEWLINE# #INDENT# ( yield listener . deferred ) #NEWLINE# #UNINDENT# #UNINDENT# except defer . CancelledError : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# self . clock . cancel_call_later ( timer , ignore_errs = True ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# current_token = user_stream . current_token #NEWLINE# result = ( yield callback ( appservice , current_token ) ) #NEWLINE# #UNINDENT# defer . returnValue ( result )