code: 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);
      })
    })
  }
docs: process documents from the stream