#NEWLINE# def process_elements( self , state , windowed_values , unused_output_watermark ) : #NEWLINE# #INDENT# if isinstance ( windowed_values , list ) : #NEWLINE# #INDENT# unwindowed = [ wv . value for wv in windowed_values ] #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# class UnwindowedValues ( observable . ObservableMixin ) : #NEWLINE# #INDENT# def __iter__( self ) : #NEWLINE# #INDENT# for wv in windowed_values : #NEWLINE# #INDENT# unwindowed_value = wv . value #NEWLINE# self . notify_observers ( unwindowed_value ) #NEWLINE# ( yield unwindowed_value ) #NEWLINE# #UNINDENT# #UNINDENT# def __repr__( self ) : #NEWLINE# #INDENT# return ( '<UnwindowedValues of %s>' % windowed_values ) #NEWLINE# #UNINDENT# #UNINDENT# unwindowed = UnwindowedValues ( ) #NEWLINE# #UNINDENT# ( yield WindowedValue ( unwindowed , MIN_TIMESTAMP , unused_output_watermark . GLOBAL_WINDOW_TUPLE ) )