Read the data from the stream.

@param {object} stream
@param {string} encoding
@param {number} length
@param {number} limit
@param {function} callback
@public
165498
function readStream (stream, encoding, length, limit, callback) {
  var complete = false
  var sync = true

  // check the length and limit options.
  // note: we intentionally leave the stream paused,
  // so users should handle the stream themselves.
  if (limit !== null && length !== null && length > limit) {
    return done(createError(413, 'request entity too large', {
      expected: length,
      length: length,
      limit: limit,
      type: 'entity.too.large'
    }))
  }

  // streams1: assert request encoding is buffer.
  // streams2+: assert the stream encoding is buffer.
  //   stream._decoder: streams1
  //   state.encoding: streams2
  //   state.decoder: streams2, specifically < 0.10.6
  var state = stream._readableState
  if (stream._decoder || (state && (state.encoding || state.decoder))) {
    // developer error
    return done(createError(500, 'stream encoding should not be set', {
      type: 'stream.encoding.set'
    }))
  }

  var received = 0
  var decoder

  try {
    decoder = getDecoder(encoding)
  } catch (err) {
    return done(err)
  }

  var buffer = decoder
    ? ''
    : []

  // attach listeners
  stream.on('aborted', onAborted)
  stream.on('close', cleanup)
  stream.on('data', onData)
  stream.on('end', onEnd)
  stream.on('error', onEnd)

  // mark sync section complete
  sync = false

  function done () {
    var args = new Array(arguments.length)

    // copy arguments
    for (var i = 0; i < args.length; i++) {
      args[i] = arguments[i]
    }

    // mark complete
    complete = true

    if (sync) {
      process.nextTick(invokeCallback)
    } else {
      invokeCallback()
    }

    function invokeCallback () {
      cleanup()

      if (args[0]) {
        // halt the stream on error
        halt(stream)
      }

      callback.apply(null, args)
    }
  }

  function onAborted () {
    if (complete) return

    done(createError(400, 'request aborted', {
      code: 'ECONNABORTED',
      expected: length,
      length: length,
      received: received,
      type: 'request.aborted'
    }))
  }

  function onData (chunk) {
    if (complete) return

    received += chunk.length

    if (limit !== null && received > limit) {
      done(createError(413, 'request entity too large', {
        limit: limit,
        received: received,
        type: 'entity.too.large'
      }))
    } else if (decoder) {
      buffer += decoder.write(chunk)
    } else {
      buffer.push(chunk)
    }
  }

  function onEnd (err) {
    if (complete) return
    if (err) return done(err)

    if (length !== null && received !== length) {
      done(createError(400, 'request size did not match content length', {
        expected: length,
        length: length,
        received: received,
        type: 'request.size.invalid'
      }))
    } else {
      var string = decoder
        ? buffer + (decoder.end() || '')
        : Buffer.concat(buffer)
      done(null, string)
    }
  }

  function cleanup () {
    buffer = null

    stream.removeListener('aborted', onAborted)
    stream.removeListener('data', onData)
    stream.removeListener('end', onEnd)
    stream.removeListener('error', onEnd)
    stream.removeListener('close', cleanup)
  }
}
335531
function (callback) {
	return function (res) {
		var resData = "";
		res.setEncoding('utf8');
		res.on('data',function(chunk) { resData+=chunk; });
		res.on('end', function() { callback(res,resData); });
	};
}
829177
function _write (chunk, encoding, callback) {
  /* istanbul ignore else */
  if (Array.isArray(this.options.input.buffer)) {
    /* istanbul ignore else */
    if (is.buffer(chunk)) {
      if (this.options.input.buffer.length === 0) {
        const that = this;
        this.on('finish', function () {
          that.streamInFinished = true;
        });
      }
      this.options.input.buffer.push(chunk);
      callback();
    } else {
      callback(new Error('Non-Buffer data on Writable Stream'));
    }
  } else {
    callback(new Error('Unexpected data on Writable Stream'));
  }
}
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);
        });
    });
}
571496
function readLines(path, callback) {
    return new Promise(function(resolve, reject) {
        const stream = fsExtra.createReadStream(path);
        stream.on('error', reject);
        stream.on('end', resolve);

        const lineReader = readline.createInterface({
            input : stream
        });
        lineReader.on('line', callback);
    });
}
1324840
function streamToString(callback) {
  var contents = '';

  return through(
    function (chunk, enc, next) {
      contents += chunk.toString('utf8');
      next();
    },
    function (next) { // flush function
      callback.bind(this)(contents, next);
    }
  );
}
432109
function(dbOrUri, stream, callback) {
    var db;
    if (arguments.length === 2) {
      callback = stream;
      stream = undefined;
    }
    if (!stream) {
      stream = process.stdin;
    }

    var state = 'start';
    var version;
    var collection;
    var endOfCollection;
    var insertQueue = [];
    var insertQueueSize = 0;

    // State machine

    var processors = {
      start: function(item, callback) {
        try {
          item = BSON.deserialize(item);
        } catch (e) {
          return callback(e);
        }
        if (typeof(item) !== 'object') {
          return callback(new Error('First BSON element is not an object'));
        }
        if (item.type !== 'mongo-dump-stream') {
          return callback(new Error('type property is not mongo-dump-stream'));
        }
        version = item.version;
        if (item.version > 2) {
          return callback(new Error('Incoming mongo-dump-stream is of a newer version, bailing out'));
        }
        endOfCollection = item.endOfCollection;
        state = 'collection';
        return callback(null);
      },
      collection: function(item, callback) {
        try {
          item = BSON.deserialize(item);
        } catch (e) {
          return callback(e);
        }
        if (item.type === 'endDatabase') {
          state = 'done';
          return callback(null);
        }
        if (item.type !== 'collection') {
          return callback(new Error('collection expected'));
        }
        var name = item.name;
        var indexes = item.indexes;
        async.series({
          create: function(callback) {
            return db.collection(name, function(err, _collection) {
              if (err) {
                return callback(err);
              }
              collection = _collection;
              return callback(null);
            });
          },
          indexes: function(callback) {
            return async.eachSeries(item.indexes, function(item, callback) {
              if (item.name === '_id_') {
                // automatic
                return setImmediate(callback);
              }
              return collection.ensureIndex(item.key, _.omit(item, 'ns', 'key'), callback);
            }, callback);
          }
        }, function(err) {
          if (err) {
            return callback(err);
          }
          state = 'documents';
          return callback(null);
        });
      },
      documents: function(item, callback) {
        if (version === '1') {
          item = BSON.deserialize(item);
          if (item.type === 'endCollection') {
            state = 'collection';
            return setImmediate(callback);
          }
          if (item.type !== 'document') {
            return callback(new Error('Document expected'));
          }
          return collection.insert(item.document, callback);
        }

        // Just scan for the unique random string that
        // appears in the end-of-collection object. No need
        // to waste time parsing BSON for this. Also
        // leverage the fact that it won't ever be
        // part of a large object to avoid scanning them all
        if (item.length < 100) {
          if (buffertools.indexOf(item, endOfCollection) !== -1) {
            return flush(function(err) {
              if (err) {
                return callback(err);
              }
              state = 'collection';
              return callback(null);
            });
          }
        }
        return async.series({
          flushIfNeeded: function(callback) {
            if (insertQueueSize + item.length <= 16777216) {
              return setImmediate(callback);
            }
            return flush(callback);
          },
          insertInQueue: function(callback) {
            insertQueueSize += item.length;
            insertQueue.push(item);
            return setImmediate(callback);
          }
        }, callback);

        function flush(callback) {
          if (!insertQueue.length) {
            // MongoDB considers it an error to be asked
            // to insert 0 documents ):
            return setImmediate(callback);
          }
          // watch out for race condition
          var insert = insertQueue;
          insertQueue = [];
          insertQueueSize = 0;
          return collection.insert(insert, function(err) {
            insertQueue = [];
            return callback(err);
          });
        }
      }
    };

    return async.series({
      connect: function(callback) {
        if (typeof(dbOrUri) !== 'string') {
          // Already a mongodb connection
          db = dbOrUri;
          return setImmediate(callback);
        }
        return mongodb.MongoClient.connect(dbOrUri, function(err, _db) {
          if (err) {
            return callback(err);
          }
          db = _db;
          return callback(null);
        });
      },
      dropCollections: function(callback) {
        return db.collections(function(err, collections) {
          if (err) {
            return callback(err);
          }
          return async.eachSeries(collections, function(item, callback) {
            if (item.collectionName.match(/^system\./)) {
              return setImmediate(callback);
            }
            return item.drop(callback);
          }, callback);
        });
      },
      loadCollections: function(finalCallback) {
        return reader({ raw: true, from: stream }, function(doc, callback) {
          return processors[state](doc, function(err) {
            if (err) {
              return callback(err);
            }
            if (state === 'done') {
              return finalCallback(null);
            }
            return callback(null);
          });
        }, function(err) {
          if (err) {
            return finalCallback(err);
          }
          return finalCallback(new Error('Premature end of stream'));
        });
      }
    }, callback);
  }
621275
def decode(self, stream, encoding='utf-8'):
        
        data = stream.read().decode(encoding).strip("\n")
        if data != "":
            logger.debug(('{0}: decoded "{1}" with encoding '
                          '{2}'.format(self.target_address, data, encoding)))
        return data
1225015
function readNext() {
    let stream, state;

    while (!isDone) {
      if (!state1.ended
          && (state2.ended || state1.totalDataLen <= state2.totalDataLen)) {
        stream = stream1;
        state = state1;
      } else if (!state2.ended) {
        stream = stream2;
        state = state2;
      } else {
        debug('All streams have ended.  No further reads.');
        return;
      }

      const data = stream.read();
      if (data === null) {
        debug(`Waiting for ${streamName(stream)} to be readable...`);
        stream.once('readable', readNext);
        return;
      }

      handleData.call(stream, state, data);
    }
  }
668005
function _readFile(fileName, callback) {
    fs.readFile(fileName, "utf8", function (err, data) {
        var content = "";
        if (!err) {
            content = data;
        }
        callback.apply(null, [fileName, content]);
    });
}
1011752
function readIntoBuf(stream) {
  return new Promise((resolve, reject) => {
    const chunks = []

    stream.on('data', chunk => {
      chunks.push(chunk)
    })

    stream.on('error', err => {
      reject(err)
    })

    stream.on('end', () => {
      resolve(Buffer.concat(chunks))
    })
  })
}
116252
function Reader(buffer, offset, length, littleEndian) {

            var bufferLength = buffer.byteLength || buffer.length;

            this.offset = offset || 0;
            this.length = length || (bufferLength - this.offset);
            this.littleEndian = littleEndian || false;

            this.view = new DataView(buffer, this.offset, this.length);
        }
980896
function readFile (file, options) {
    // return new Promise that will be resolved with data
    return new Promise((resolve, reject) => {
        // create write stream
        this.createReadStream(file, options).then(stream => {
            // create new concat stream that will call back with stream data
            var concatStream = ConcatStream(buffer => {
                // resolve with string if encoding set or buffer
                var data = defined(options) && defined(options.encoding)
                    ? buffer.toString(options.encoding)
                    : buffer
                // resolve with data
                resolve(data)
            })
            // reject on errors
            stream.on('error', reject)
            // pipe read stream to concat stream which will resolve when done
            stream.pipe(concatStream)
        })
        // pass rejection
        .catch(reject)
    })
}
593840
function write (chunk, encoding, fn) {
  if (!this._parserInit) init(this);
  debug('write(%o bytes)', chunk.length);

  // XXX: old Writable stream API compat... remove at some point...
  if ('function' == typeof encoding) fn = encoding;

  data(this, chunk, null, fn);
}
988305
function logReadableStream(stream) {
        var encoding = "utf8";

        stream.setEncoding(encoding);

        stream.on("data", function (chunk) {
            writeToLog(chunk, encoding);
        });
    }
874490
function startStream () {
  process.stdin.setEncoding('utf8')
  process.stdin.on('data', function(data) {
    return {
      "--parse"     : function (string) {
        var result = parse(string)
        process.stdout.write(JSON.stringify(result, null, 2))
      },
      "--serialize" : function (string) {
        var result = serialize(JSON.parse(string))
        process.stdout.write(result)
      }
    }[$mode](data)
  })
}
857047
function consumeStream(stream, callback) {
    var chunks = [];
    stream.on("data", onData);
    stream.on("end", onEnd);
    stream.on("error", onError);
    function onData(chunk) {
        chunks.push(chunk);
    }
    function onEnd() {
        cleanup();
        callback(null, chunks.join(""));
    }
    function onError(err) {
        cleanup();
        callback(err);
    }
    function cleanup() {
        stream.removeListener("data", onData);
        stream.removeListener("end", onEnd);
        stream.removeListener("error", onError);
    }
}