#NEWLINE# def process_timer( self , window_id , unused_name , time_domain , timestamp , state ) : #NEWLINE# #INDENT# if self . is_merging : #NEWLINE# #INDENT# state = MergeableStateAdapter ( state ) #NEWLINE# #UNINDENT# window = state . get_window ( window_id ) #NEWLINE# if state . get_state ( window , self . TOMBSTONE ) : #NEWLINE# #INDENT# return #NEWLINE# #UNINDENT# if ( time_domain == TimeDomain . WATERMARK ) : #NEWLINE# #INDENT# if ( ( not self . is_merging ) or ( window in state . known_windows ( ) ) ) : #NEWLINE# #INDENT# context = state . at ( window ) #NEWLINE# if self . trigger_fn . should_fire ( timestamp , window , context ) : #NEWLINE# #INDENT# finished = self . trigger_fn . on_fire ( timestamp , window , context ) #NEWLINE# ( yield self . _output ( window , window , state ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# raise Exception ( ( 'Unexpected time domain: %s' % time_domain ) )