def handle_pipes(self):
    'Use pipes to determine job groups. Check if every pipe has exactly\n        one consumer'
    for job in self.needrun_jobs:
        candidate_groups = set()
        if (job.group is not None):
            candidate_groups.add(job.group)
        all_depending = set()
        has_pipe = False
        for f in job.output:
            if is_flagged(f, 'pipe'):
                if job.is_run:
                    raise WorkflowError("Rule defines pipe output but uses a 'run' directive. This is not possible for technical reasons. Consider using 'shell' or 'script'.", rule=job.rule)
                has_pipe = True
                depending = [j for (j, files) in self.depending[job].items() if (f in files)]
                if (len(depending) > 1):
                    raise WorkflowError('Output file {} is marked as pipe but more than one job depends on it. Make sure that any pipe output is only consumed by one job'.format(f), rule=job.rule)
                elif (len(depending) == 0):
                    raise WorkflowError('Output file {} is marked as pipe but it has no consumer. This is invalid because it can lead to a dead lock.'.format(f), rule=job.rule)
                depending = depending[0]
                if depending.is_run:
                    raise WorkflowError("Rule consumes pipe input but uses a 'run' directive. This is not possible for technical reasons. Consider using 'shell' or 'script'.", rule=job.rule)
                all_depending.add(depending)
                if (depending.group is not None):
                    candidate_groups.add(depending.group)
        if (not has_pipe):
            continue
        if (len(candidate_groups) > 1):
            raise WorkflowError('An output file is marked as pipe, but consuming jobs are part of conflicting groups.', rule=job.rule)
        elif candidate_groups:
            group = candidate_groups.pop()
        else:
            group = str(uuid.uuid4())
        job.group = group
        for j in all_depending:
            j.group = group
Use pipes to determine job groups. Check if every pipe has exactly
one consumer
def handle_pipes(self):
    'Use pipes to determine job groups. Check if every pipe has exactly\n        one consumer'
    for job in self.needrun_jobs:
        candidate_groups = set()
        if (job.group is not None):
            candidate_groups.add(job.group)
        all_depending = set()
        has_pipe = False
        for f in job.output:
            if is_flagged(f, 'pipe'):
                if job.is_run:
                    raise WorkflowError("Rule defines pipe output but uses a 'run' directive. This is not possible for technical reasons. Consider using 'shell' or 'script'.", rule=job.rule)
                has_pipe = True
                depending = [j for (j, files) in self.depending[job].items() if (f in files)]
                if (len(depending) > 1):
                    raise WorkflowError('Output file {} is marked as pipe but more than one job depends on it. Make sure that any pipe output is only consumed by one job'.format(f), rule=job.rule)
                elif (len(depending) == 0):
                    raise WorkflowError('Output file {} is marked as pipe but it has no consumer. This is invalid because it can lead to a dead lock.'.format(f), rule=job.rule)
                depending = depending[0]
                if depending.is_run:
                    raise WorkflowError("Rule consumes pipe input but uses a 'run' directive. This is not possible for technical reasons. Consider using 'shell' or 'script'.", rule=job.rule)
                all_depending.add(depending)
                if (depending.group is not None):
                    candidate_groups.add(depending.group)
        if (not has_pipe):
            continue
        if (len(candidate_groups) > 1):
            raise WorkflowError('An output file is marked as pipe, but consuming jobs are part of conflicting groups.', rule=job.rule)
        elif candidate_groups:
            group = candidate_groups.pop()
        else:
            group = str(uuid.uuid4())
        job.group = group
        for j in all_depending:
            j.group = group