Pipe several transformers end to end.
508234
def pipes(stream, *transformers):
    """"""
    for transformer in transformers:
        stream = stream.pipe(transformer)
    return stream
616587
function concatStream(list) {
  list = list.map(toStream);
  const transform = transformWithCancel(async function(reason) {
    await Promise.all(transforms.map(stream => cancel(stream, reason)));
  });
  let prev = Promise.resolve();
  const transforms = list.map((stream, i) => transformPair(stream, (readable, writable) => {
    prev = prev.then(() => pipe(readable, transform.writable, {
      preventClose: i !== list.length - 1
    }));
    return prev;
  }));
  return transform.readable;
}
1176346
def stream_lines(tokens, line_num)
      yield line_start(line_num)

      tokens.each do |token, value|
        yield @inner.span(token, value)
      end

      yield line_end(line_num)
    end
1325549
def convert(self, verbose=True):
        
        input_transformer = self._input_class(self._filename)
        output_transformer = self._output_class()

        containers = input_transformer.ingest_containers()

        output_containers = []

        for container in containers:
            converted_container = self._convert_container(
                container,
                input_transformer,
                output_transformer
            )

            validated = output_transformer.validate(converted_container)

            output_containers.append(validated)

        return output_transformer.emit_containers(output_containers, verbose)
1077750
async def chain(*sources):
    
    for source in sources:
        async with streamcontext(source) as streamer:
            async for item in streamer:
                yield item
1282018
def _reduce_input(self, inputs, reducer, final=NotImplemented):
        
        for key, values in groupby(inputs, key=lambda x: self.internal_serialize(x[0])):
            for output in reducer(self.deserialize(key), (v[1] for v in values)):
                yield output
        if final != NotImplemented:
            for output in final():
                yield output
        self._flush_batch_incr_counter()
77730
def _nodes(output, parent=None):
    """"""
    # NOTE refactor so all outputs behave the same
    entity = getattr(output, 'entity', output)

    if isinstance(entity, Collection):
        for member in entity.members:
            if parent is not None:
                member = attr.evolve(member, parent=parent)
            yield from _nodes(member)
        yield output
    else:
        yield output
472723
def alternator(*pipes):
    '''  '''
    try:
        for p in cycle(map(iter, pipes)):
            yield next(p)
    except StopIteration:
        pass
773073
def to_pipeline
      pipeline = []
      pipeline.push({ "$skip" => skip }) if skip
      pipeline.push({ "$limit" => limit }) if limit
      pipeline.push({ "$sort" => sort }) if sort
      pipeline
    end
266665
def ends(self, layer):
        """"""
        ends = []
        for data in self[layer]:
            ends.append(data[END])
        return ends
540553
function pipeFileArray (arr) {
    testPipeFile();

    // Just end WriteStream if no files need to be read to keep default pipe behavior.
    if (arr.length === 0) {
        writable.end();
    } else {
        iterateData.collection = arr;

        iteratePipeFile();
    }
}
1003643
public static <E> void pipe(Iterator<E> iterator, OutputIterator<E> outputIterator) {
        new ConsumeIntoOutputIterator<>(outputIterator).apply(iterator);
    }
1360667
def end_output (self, **kwargs):
        """"""
        self.write_edges()
        self.end_graph()
        if self.has_part("outro"):
            self.write_outro()
        self.close_fileoutput()
156275
def pipeline(names, steps):
    """"""
    steps, times = zip(*map(_maybe_timed, steps))
    fit_time = sum(times)
    if any(s is FIT_FAILURE for s in steps):
        fit_est = FIT_FAILURE
    else:
        fit_est = Pipeline(list(zip(names, steps)))
    return fit_est, fit_time
527963
def complete_shells(line, text, predicate=lambda i: True):
    """"""
    res = [i.display_name + ' ' for i in dispatchers.all_instances() if
           i.display_name.startswith(text) and
           predicate(i) and
           ' ' + i.display_name + ' ' not in line]
    return res
533796
def skip_first(pipe, items=1):
    ''' 
    '''
    pipe = iter(pipe)
    for i in skip(pipe, items):
        yield i
880411
function transformer(data) {
        var newData = {};
        var keys = Object.keys(data), length = keys.length, key, i;
        for ( i = 0; i < length; i++) {
          key = keys[i];
          newData[key] = data[key];
        }
        return newData;
      }