#NEWLINE# def _ScrubFanIn( self , input_stages , output_stage , sink_generator ) : #NEWLINE# #INDENT# "Wires one or more stages' sinks to one stage's sources.\n\n    Args:\n      input_stages: A list of stages.\n      output_stage: The next stage.\n      sink_generator: Generates temporary storage links.\n\n    Raises:\n      PipelineError: Invalid definition.\n    " #NEWLINE# logging . info ( 'scrubbing stages Fan In:\n%s\n->\n%s\n' , json . dumps ( input_stages , indent = 2 , separators = ( ',' , ': ' ) ) , json . dumps ( output_stage , indent = 2 , separators = ( ',' , ': ' ) ) ) #NEWLINE# if ( 'sources' not in output_stage ) : #NEWLINE# #INDENT# output_stage [ 'sources' ] = [ ] #NEWLINE# #UNINDENT# for stage in input_stages : #NEWLINE# #INDENT# if ( ( 'sinks' in stage ) and sink_generator [ 'sinks' ] ) : #NEWLINE# #INDENT# for sink in stage [ 'sinks' ] : #NEWLINE# #INDENT# if ( sink not in output_stage [ 'sources' ] ) : #NEWLINE# #INDENT# output_stage [ 'sources' ] . append ( sink ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# link = sink_generator ( ) #NEWLINE# stage [ 'sinks' ] = [ link ] #NEWLINE# output_stage [ 'sources' ] . append ( link )