def _ScrubFanIn(self, input_stages, output_stage, sink_generator):
    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=(',', ': ')))
    if 'sources' not in output_stage:
      output_stage['sources'] = []

    # Write any input stage's sink as a source on the output stage,
    # if none exist, generate a temporary link and bind together.
    for stage in input_stages:
      if 'sinks' in stage and stage['sinks']:
        for sink in stage['sinks']:
          if sink not in output_stage['sources']:
            output_stage['sources'].append(sink)
      else:
        link = sink_generator()
        stage['sinks'] = [link]
        output_stage['sources'].append(link)
Return position of next evaluated element
dataset/ETHPy150Open GoogleCloudPlatform/Data-Pipeline/app/src/pipelines/runner.py/PipelineRunner._ScrubFanIn--crossover with--dataset/ETHPy150Open kachayev/fn.py/fn/stream.py/Stream.cursor