lib/index.js
function read (stream, opts, db, cb) {
  var schema = new exports.Schema;

  if (opts.onType) {
    schema.onType = opts.onType;
  }

  stream.once('error', function (err) {
    debug('aborting with error: ' + err);
    db.close(cb.bind(err));
  });

  stream.on('data', function (doc) {
    var err = schema.consume(doc);
    if (err) return abort(err, stream, db, cb);
  })

  stream.on('end', function () {
    debug('db closing');
    db.close(function () {
      debug('db closed');
      cb(null, schema);
    })
  })
}
function read (stream, opts, db, cb) {
  var schema = new exports.Schema;

  if (opts.onType) {
    schema.onType = opts.onType;
  }

  stream.once('error', function (err) {
    debug('aborting with error: ' + err);
    db.close(cb.bind(err));
  });

  stream.on('data', function (doc) {
    var err = schema.consume(doc);
    if (err) return abort(err, stream, db, cb);
  })

  stream.on('end', function () {
    debug('db closing');
    db.close(function () {
      debug('db closed');
      cb(null, schema);
    })
  })
}
function read ( stream , opts , db , cb ) { var schema = new exports . Schema ; if ( opts . onType ) { schema . onType = opts . onType ; } stream . once ( 'error' , function ( err ) { debug ( 'aborting with error: ' + err ) ; db . close ( cb . bind ( err ) ) ; } ) ; stream . on ( 'data' , function ( doc ) { var err = schema . consume ( doc ) ; if ( err ) return abort ( err , stream , db , cb ) ; } ) stream . on ( 'end' , function ( ) { debug ( 'db closing' ) ; db . close ( function ( ) { debug ( 'db closed' ) ; cb ( null , schema ) ; } ) } ) }
process documents from the stream
process documents from the stream