Leverage x streams to transform, delayed echoing in this example, data from our readable stream.
169093
function run (x, cb) {
  const s = speculum(null, Echo, x)
  s.on('end', cb)
  s.on('error', cb)

  const reader = new Count({ highWaterMark: 0 }, 10)

  reader.pipe(s).resume()
}
971338
function EmitContinuation (stream) {

  stream._processing++;

  var self = function (err, data) {
    if (err)
      self.err = err;
    else if (Array.isArray(data))
      self.data = data;
    else
      self.data = [data];

    self.done = true;

    stream._processing--;

    stream._next();
  };
  return self;
}
54540
def _transform_data(self, X):
        """"""

        if self._binarizers == []:
            raise NotFittedError()

        if self.binarize is not None:
            X = binarize(X, threshold=self.binarize)

        if len(self._binarizers) != X.shape[1]:
            raise ValueError(
                "Expected input with %d features, got %d instead" %
                (len(self._binarizers), X.shape[1]))

        X_parts = []

        for i in range(X.shape[1]):

            X_i = self._binarizers[i].transform(X[:, i])

            # sklearn returns ndarray with shape (samples, 1) on binary input.
            if self._binarizers[i].classes_.shape[0] == 2:
                X_parts.append(1 - X_i)

            X_parts.append(X_i)

        return np.concatenate(X_parts, axis=1)
356419
function streamToText(stream) {
    return new Promise((fulfill, reject) => {
        let chunks = [];
        let length = 0;
        stream.setEncoding('utf8');
        stream.on('data', chunk => {
            length += chunk.length;
            if (length > MAX_TEXT_LENGTH) {
                reject(new Error('Response is too big'));
                return;
            }
            chunks.push(chunk);
        });
        stream.on('end', () => {
            fulfill(chunks.join(''));
        });
        stream.on('error', err => {
            reject(err);
        });
    });
}
751519
function transformArrayData(stream, item, cb) {
    var row = [];
    if (stream.totalCount++) {
        row.push(stream.rowDelimiter);
    }
    row.push(stream.formatter(item));
    return row.join("");
}
414844
function transform (chunk, output, fn) {
  if (!this._parserInit) init(this);
  debug('transform(%o bytes)', chunk.length);

  // XXX: old Transform stream API compat... remove at some point...
  if ('function' != typeof output) {
    output = this._parserOutput;
  }

  data(this, chunk, output, fn);
}
182593
async function convertStream(data, streaming) {
  if (!streaming && util.isStream(data)) {
    return stream.readToEnd(data);
  }
  if (streaming && !util.isStream(data)) {
    data = new ReadableStream({
      start(controller) {
        controller.enqueue(data);
        controller.close();
      }
    });
  }
  if (streaming === 'node') {
    data = stream.webToNode(data);
  }
  return data;
}
183042
def transform(self, X, y=None):
        '''
        
        '''
        return Parallel(n_jobs=self.n_jobs)(delayed(self._transform)(x)
                                            for x in X)
1161990
function streamToValue (response, callback) {
  pump(
    response,
    concat((data) => callback(null, data)),
    (err) => {
      if (err) {
        callback(err)
      }
    })
}
335457
def parse_streams(self):
        
        logger.debug("Parsing streams of {}".format(self.input_file))

        cmd = [
            self.ffmpeg_normalize.ffmpeg_exe, '-i', self.input_file,
            '-c', 'copy', '-t', '0', '-map', '0',
            '-f', 'null', NUL
        ]

        cmd_runner = CommandRunner(cmd)
        cmd_runner.run_command()
        output = cmd_runner.get_output()

        logger.debug("Stream parsing command output:")
        logger.debug(output)

        output_lines = [line.strip() for line in output.split('\n')]

        for line in output_lines:

            if not line.startswith('Stream'):
                continue

            stream_id_match = re.search(r'#0:([\d]+)', line)
            if stream_id_match:
                stream_id = int(stream_id_match.group(1))
                if stream_id in self._stream_ids():
                    continue
            else:
                continue

            if 'Audio' in line:
                logger.debug("Found audio stream at index {}".format(stream_id))
                sample_rate_match = re.search(r'(\d+) Hz', line)
                sample_rate = int(sample_rate_match.group(1)) if sample_rate_match else None
                bit_depth_match = re.search(r's(\d+)p?,', line)
                bit_depth = int(bit_depth_match.group(1)) if bit_depth_match else None
                self.streams['audio'][stream_id] = AudioStream(self, stream_id, sample_rate, bit_depth)

            elif 'Video' in line:
                logger.debug("Found video stream at index {}".format(stream_id))
                self.streams['video'][stream_id] = VideoStream(self, stream_id)

            elif 'Subtitle' in line:
                logger.debug("Found subtitle stream at index {}".format(stream_id))
                self.streams['subtitle'][stream_id] = SubtitleStream(self, stream_id)

        if not self.streams['audio']:
            raise FFmpegNormalizeError(
                "Input file {} does not contain any audio streams"
                .format(self.input_file))

        if os.path.splitext(self.output_file)[1].lower() in ['.wav', '.mp3', '.aac']:
            logger.warning(
                "Output file only supports one stream. "
                "Keeping only first audio stream."
            )
            first_stream = list(self.streams['audio'].values())[0]
            self.streams['audio'] = {first_stream.stream_id: first_stream}
            self.streams['video'] = {}
            self.streams['subtitle'] = {}
581391
def _xread(self, streams, timeout=0, count=None, latest_ids=None):
        
        if latest_ids is None:
            latest_ids = ['$'] * len(streams)
        if len(streams) != len(latest_ids):
            raise ValueError(
                'The streams and latest_ids parameters must be of the '
                'same length'
            )

        count_args = [b'COUNT', count] if count else []
        if timeout is None:
            block_args = []
        elif not isinstance(timeout, int):
            raise TypeError(
                "timeout argument must be int, not {!r}".format(timeout))
        else:
            block_args = [b'BLOCK', timeout]
        return block_args + count_args + [b'STREAMS'] + streams + latest_ids
924213
def load_stream(self, stream):
        
        # load the batches
        for batch in self.serializer.load_stream(stream):
            yield batch

        # load the batch order indices
        num = read_int(stream)
        batch_order = []
        for i in xrange(num):
            index = read_int(stream)
            batch_order.append(index)
        yield batch_order
463963
def echo(transferred, toBeTransferred, suffix=''):
        ''' 
        '''
        bar_len = 60                
        rate = transferred/float(toBeTransferred)
        
        filled_len = int(round(bar_len * rate))
        _percents = "%s%s" %(round(100.0 * rate, 1), "%")
        
        end_str = "\r"
        _bar = '=' * filled_len + '-' * (bar_len - filled_len)
        print("[%s] %s ...%s%s" %(_bar, _percents, suffix, end_str))
1191496
function toStream(input) {
  let streamType = isStream(input);
  if (streamType === 'node') {
    return nodeToWeb(input);
  } else if (streamType) {
    return input;
  }
  return new ReadableStream({
    start(controller) {
      controller.enqueue(input);
      controller.close();
    }
  });
}
67545
function ZXYStream(source, options) {
    if (!source) throw new TypeError('MBTiles source required');

    options = options || {};

    this.source = source;
    this.batch = options.batch || 1000;

    stream.Readable.call(this);
}
404511
function Transformer( file, res ) {
			stream.Transform.call( this );

			this.file = file;
			this.res = res;
			this.chunks = [];
		}
1321725
void transformAndQueue(T event, long systemCurrentTimeMillis) {
        // if you're super unlucky, this blows up the stack
        if (rand.nextDouble() < 0.05) {
            // duplicate this message (note recursion means maybe more than duped)
            transformAndQueue(event, systemCurrentTimeMillis);
        }

        long delayms = nextZipfDelay();
        delayed.add(systemCurrentTimeMillis + delayms, event);
    }