ClassicReadable wraps a "classic" readable stream.

@class ClassicReadable
@constructor
@extends ZReadable
@uses _Classic
@param {Stream} stream - The classic stream being wrapped
@param {Object} [options] - Stream options
502991
function ClassicReadable(stream, options) {
	Readable.call(this, options);
	classicMixins.call(this, stream, options);

	// Readable streams already include a wrapping for Classic Streams
	this.wrap(stream);
}
175860
function BlobReadStream(blob, options) {
  if (!(this instanceof BlobReadStream)) {
    return new BlobReadStream(blob, options);
  }

  Readable.call(this, options);

  options = options || {};
  this.blob = blob;
  this.slice = blob.slice || blob.webkitSlice || blob.mozSlice;
  this.start = 0;
  this.sync = options.synchronous || false;

  var fileReader;

  if (options.synchronous) {
    fileReader = this.fileReader = new FileReaderSync();
  } else {
    fileReader = this.fileReader = new FileReader();
  }

  fileReader.onload = bind(this, '_onload');
  fileReader.onerror = bind(this, '_onerror');
}
561008
function emitReadable(stream) {
	  var state = stream._readableState;
	  state.needReadable = false;
	  if (!state.emittedReadable) {
	    debug('emitReadable', state.flowing);
	    state.emittedReadable = true;
	    if (state.sync) nextTick(emitReadable_, stream);else emitReadable_(stream);
	  }
	}
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;
}
1158600
function(stream, callback) {
                    var content = '';
                    stream.on('data', function(chunk) {
                      content += chunk;
                    });
                    stream.on('end', function() {
                      callback(content);
                    });
                  }
804096
function ClassicDuplex(stream, options) {
	var readable, writable, classicReadable, classicWritable, self = this;

	readable = new PassThrough();
	writable = new PassThrough();
	CompoundDuplex.call(self, writable, readable, options);
	classicMixins.call(this, stream, options);

	classicReadable = this._internalReadable = new ClassicReadable(stream, options);
	classicReadable.on('error', this._duplexHandleInternalError.bind(this));

	classicWritable = this._internalWritable = new ClassicWritable(stream, options);
	classicWritable.on('error', this._duplexHandleInternalError.bind(this));

	writable.pipe(classicWritable);
	classicReadable.pipe(readable);
}
1252356
function wrap(stream) {
  stream.on('error', error => {
    gutil.log(gutil.colors.red(error.message));
    gutil.log(error.stack);
    if (watching) {
      gutil.log(gutil.colors.yellow('[aborting]'));
      stream.end();
    } else {
      gutil.log(gutil.colors.yellow('[exiting]'));
      process.exit(1);
    }
    ringBell();
  });
  return stream;
}
1224407
function emitReadable(stream) {
  var state = stream._readableState;
  state.needReadable = false;
  if (state.emittedReadable)
    return;

  state.emittedReadable = true;
  if (state.sync)
    timers.setImmediate(function() {
      emitReadable_(stream);
    });
  else
    emitReadable_(stream);
}
343911
function extractNestedStreams(obj) {
  return flatten(
    Object.keys(obj).map(key => {
      // DEPRECATED I can't think of a usecase
      // if (typeof obj[key] === 'object') {
      // 	return extractNestedStreams(obj[key]);
      // }
      if (obj[key] === null || obj[key] === undefined) {
        return [];
      }
      if (isStream(obj[key])) {
        return [
          {
            parent: obj,
            key,
            stream: obj[key]
          }
        ];
      }
      if (typeof obj[key] === "object") {
        return extractNestedStreams(obj[key]);
      }
      return [];
    })
  );
}
583923
public function addStream(IStream $stream) : void
    {
        if (!$stream->isReadable()) {
            throw new InvalidArgumentException('Stream must be readable');
        }

        $this->isSeekable = $this->isSeekable && $stream->isSeekable();
        $this->streams[] = $stream;
    }
521360
function stream (input, value) {
  value.on('data', data => input.push(data))
  value.on('error', err => error(input, err))
  value.on('end', () => input.push(null))
}
224336
function(stream) {
		var exportKey = function(obj, env) {
			var keyString = keyStringGenerator(obj, env);

			if(stream.constElements) {
				if(obj.___jp_originalJsonLine)
					return keyString + separator + obj.___jp_originalJsonLine;
			} else {
				if(obj.___jp_originalJsonLine)
					delete obj.___jp_originalJsonLine;
			}

			return keyString + separator + JSON.stringify(obj);
		};

		return stream.pipe(jp.map(exportKey));
	}
701803
private function initReadableStream()
    {
        $context = stream_context_get_options($this->context);

        $this->stream = new ReadableStream(
            $context[$this->protocol]['collectionWrapper'],
            $context[$this->protocol]['file']
        );

        return true;
    }
1333418
function addToStream(stream) {
  // Don't add more than once
  if (stream.listeners('addListener').includes(addListener)) {
    return stream;
  }

  stream.on('newListener', addListener);

  return stream;
}
958222
function readableNodeToWeb(nodeStream) {
    return new ReadableStream({
        start(controller) {
            nodeStream.pause();
            nodeStream.on('data', chunk => {
                controller.enqueue(chunk);
                nodeStream.pause();
            });
            nodeStream.on('end', () => controller.close());
            nodeStream.on('error', (e) => controller.error(e));
        },
        pull(controller) {
            nodeStream.resume();
        },
        cancel(reason) {
            nodeStream.pause();
        }
    });
}
951437
function unwrap(value) {
      if (streaming) return value;
      return new Promise(function (resolve) {
        stream._read = function () {
          release();
          this._read = release;
          resolve(value);
        }
      });
    }
226123
function MuteStream (opts) {
  Stream.apply(this)
  opts = opts || {}
  this.writable = this.readable = true
  this.muted = false
  this.on('pipe', this._onpipe)
  this.replace = opts.replace

  // For readline-type situations
  // This much at the start of a line being redrawn after a ctrl char
  // is seen (such as backspace) won't be redrawn as the replacement
  this._prompt = opts.prompt || null
  this._hadControl = false
}