自学教程：C++ BSON函数代码示例

 void init() {     BSONArrayBuilder members;     members.append(BSON("_id" << 0 << "host" << "host1"));     _config = ReplSetConfig::make(BSON("_id" << "foo" << "members" << members.arr()));     _myConfig = new ReplSetConfig::MemberCfg(); }

    bool Cloner::go(const char *masterHost, string& errmsg, const string& fromdb, bool logForRepl, bool slaveOk, bool useReplAuth, bool snapshot) {		massert( "useReplAuth is not written to replication log", !useReplAuth || !logForRepl );        string todb = database->name;        stringstream a,b;        a << "localhost:" << cmdLine.port;        b << "127.0.0.1:" << cmdLine.port;        bool masterSameProcess = ( a.str() == masterHost || b.str() == masterHost );        if ( masterSameProcess ) {            if ( fromdb == todb && database->path == dbpath ) {                // guard against an "infinite" loop                /* if you are replicating, the local.sources config may be wrong if you get this */                errmsg = "can't clone from self (localhost).";                return false;            }        }        /* todo: we can put these releases inside dbclient or a dbclient specialization.           or just wait until we get rid of global lock anyway.           */        string ns = fromdb + ".system.namespaces";        list<BSONObj> toClone;        {              dbtemprelease r;		            auto_ptr<DBClientCursor> c;            {                if ( !masterSameProcess ) {                    auto_ptr< DBClientConnection > c( new DBClientConnection() );                    if ( !c->connect( masterHost, errmsg ) )                        return false;                    if( !replAuthenticate(c.get()) )                        return false;                                        conn = c;                } else {                    conn.reset( new DBDirectClient() );                }                c = conn->query( ns.c_str(), BSONObj(), 0, 0, 0, slaveOk ? Option_SlaveOk : 0 );            }            if ( c.get() == 0 ) {                errmsg = "query failed " + ns;                return false;            }                        while ( c->more() ){                BSONObj collection = c->next();                log(2) << "/t cloner got " << collection << endl;                BSONElement e = collection.findElement("name");                if ( e.eoo() ) {                    string s = "bad system.namespaces object " + collection.toString();                    massert(s.c_str(), false);                }                assert( !e.eoo() );                assert( e.type() == String );                const char *from_name = e.valuestr();                if( strstr(from_name, ".system.") ) {                     /* system.users is cloned -- but nothing else from system. */                    if( strstr(from_name, ".system.users") == 0 ){                        log(2) << "/t/t not cloning because system collection" << endl;                        continue;                    }                }                else if( strchr(from_name, '$') ) {                    // don't clone index namespaces -- we take care of those separately below.                    log(2) << "/t/t not cloning because has $ " << endl;                    continue;                }                                            toClone.push_back( collection.getOwned() );            }        }        for ( list<BSONObj>::iterator i=toClone.begin(); i != toClone.end(); i++ ){            {                dbtemprelease r;            }            BSONObj collection = *i;            log(2) << "  really will clone: " << collection << endl;            const char * from_name = collection["name"].valuestr();            BSONObj options = collection.getObjectField("options");                        /* change name "<fromdb>.collection" -> <todb>.collection */            const char *p = strchr(from_name, '.');            assert(p);            string to_name = todb + p;            //if( !options.isEmpty() )            {                string err;                const char *toname = to_name.c_str();                userCreateNS(toname, options, err, logForRepl);                /* chunks are big enough that we should create the _id index up front, that should                   be faster. perhaps we should do that for everything?  Not doing that yet -- not sure                   how we want to handle _id-less collections, and we might not want to create the index//.........这里部分代码省略.........

namespace IndexUpdateTests {    static const char* const _ns = "unittests.indexupdate";    DBDirectClient _client;    ExternalSortComparison* _aFirstSort = BtreeBasedBulkAccessMethod::getComparison(0, BSON("a" << 1));    /**     * Test fixture for a write locked test using collection _ns.  Includes functionality to     * partially construct a new IndexDetails in a manner that supports proper cleanup in     * dropCollection().     */    class IndexBuildBase {    public:        IndexBuildBase() :            _ctx( _ns ) {            _client.createCollection( _ns );        }        ~IndexBuildBase() {            _client.dropCollection( _ns );            killCurrentOp.reset();        }        Collection* collection() {            return _ctx.ctx().db()->getCollection( _ns );        }    protected:    // QUERY_MIGRATION#if 0        /** @return IndexDetails for a new index on a:1, with the info field populated. */        IndexDescriptor* addIndexWithInfo() {            BSONObj indexInfo = BSON( "v" << 1 <<                                      "key" << BSON( "a" << 1 ) <<                                      "ns" << _ns <<                                      "name" << "a_1" );            int32_t lenWHdr = indexInfo.objsize() + Record::HeaderSize;            const char* systemIndexes = "unittests.system.indexes";            DiskLoc infoLoc = allocateSpaceForANewRecord( systemIndexes,                                                          nsdetails( systemIndexes ),                                                          lenWHdr,                                                          false );            Record* infoRecord = reinterpret_cast<Record*>( getDur().writingPtr( infoLoc.rec(),                                                                                 lenWHdr ) );            memcpy( infoRecord->data(), indexInfo.objdata(), indexInfo.objsize() );            addRecordToRecListInExtent( infoRecord, infoLoc );            IndexCatalog::IndexBuildBlock blk( collection()->getIndexCatalog(), "a_1", infoLoc );            blk.success();            return collection()->getIndexCatalog()->findIndexByName( "a_1" );        }#endif        Client::WriteContext _ctx;        OperationContextImpl _txn;    };    /** addKeysToPhaseOne() adds keys from a collection's documents to an external sorter. */    // QUERY_MIGRATION#if 0    class AddKeysToPhaseOne : public IndexBuildBase {    public:        void run() {            // Add some data to the collection.            int32_t nDocs = 130;            for( int32_t i = 0; i < nDocs; ++i ) {                _client.insert( _ns, BSON( "a" << i ) );            }            IndexDescriptor* id = addIndexWithInfo();            // Create a SortPhaseOne.            SortPhaseOne phaseOne;            ProgressMeterHolder pm (cc().curop()->setMessage("AddKeysToPhaseOne",                                                             "AddKeysToPhaseOne Progress",                                                             nDocs,                                                             nDocs));            // Add keys to phaseOne.            BtreeBasedBuilder::addKeysToPhaseOne( collection(),                                                  id,                                                  BSON( "a" << 1 ),                                                  &phaseOne,                                                  pm.get(), true );            // Keys for all documents were added to phaseOne.            ASSERT_EQUALS( static_cast<uint64_t>( nDocs ), phaseOne.n );        }    };    /** addKeysToPhaseOne() aborts if the current operation is killed. */    class InterruptAddKeysToPhaseOne : public IndexBuildBase {    public:        InterruptAddKeysToPhaseOne( bool mayInterrupt ) :            _mayInterrupt( mayInterrupt ) {        }        void run() {            // It's necessary to index sufficient keys that a RARELY condition will be triggered.            int32_t nDocs = 130;            // Add some data to the collection.            for( int32_t i = 0; i < nDocs; ++i ) {                _client.insert( _ns, BSON( "a" << i ) );            }            IndexDescriptor* id = addIndexWithInfo();            // Create a SortPhaseOne.            SortPhaseOne phaseOne;//.........这里部分代码省略.........

 TEST( ExpressionParserArrayTest, SizeBad ) {     BSONObj query = BSON( "x" << BSON( "$size" << BSONNULL ) );     StatusWithExpression result = ExpressionParser::parse( query );     ASSERT_FALSE( result.isOK() ); }

    StatusWith<string> isValidShard(const string& name,                                    const ConnectionString& shardConnectionString,                                    ScopedDbConnection& conn) {        if (conn->type() == ConnectionString::SYNC) {            return Status(ErrorCodes::BadValue,                          "can't use sync cluster as a shard; for a replica set, "                          "you have to use <setname>/<server1>,<server2>,...");        }        BSONObj resIsMongos;        // (ok == 0) implies that it is a mongos        if (conn->runCommand("admin", BSON("isdbgrid" << 1), resIsMongos)) {            return Status(ErrorCodes::BadValue,                          "can't add a mongos process as a shard");        }        BSONObj resIsMaster;        if (!conn->runCommand("admin", BSON("isMaster" << 1), resIsMaster)) {            return Status(ErrorCodes::OperationFailed,                          str::stream() << "failed running isMaster: " << resIsMaster);        }        // if the shard has only one host, make sure it is not part of a replica set        string setName = resIsMaster["setName"].str();        string commandSetName = shardConnectionString.getSetName();        if (commandSetName.empty() && !setName.empty()) {            return Status(ErrorCodes::BadValue,                          str::stream() << "host is part of set " << setName << "; "                                        << "use replica set url format "                                        << "<setname>/<server1>,<server2>, ...");        }        if (!commandSetName.empty() && setName.empty()) {            return Status(ErrorCodes::OperationFailed,                          str::stream() << "host did not return a set name; "                                        << "is the replica set still initializing? "                                        << resIsMaster);        }        // if the shard is part of replica set, make sure it is the right one        if (!commandSetName.empty() && (commandSetName != setName)) {            return Status(ErrorCodes::OperationFailed,                          str::stream() << "host is part of a different set: " << setName);        }        if (setName.empty()) {            // check this isn't a --configsvr            BSONObj res;            bool ok = conn->runCommand("admin",                                       BSON("replSetGetStatus" << 1),                                       res);            if(!ok && res["info"].type() == String && res["info"].String() == "configsvr") {                return Status(ErrorCodes::BadValue,                              "the specified mongod is a --configsvr and "                              "should thus not be a shard server");            }        }        // if the shard is part of a replica set,        // make sure all the hosts mentioned in 'shardConnectionString' are part of        // the set. It is fine if not all members of the set are present in 'shardConnectionString'.        bool foundAll = true;        string offendingHost;        if (!commandSetName.empty()) {            set<string> hostSet;            BSONObjIterator iter(resIsMaster["hosts"].Obj());            while (iter.more()) {                hostSet.insert(iter.next().String()); // host:port            }            if (resIsMaster["passives"].isABSONObj()) {                BSONObjIterator piter(resIsMaster["passives"].Obj());                while (piter.more()) {                    hostSet.insert(piter.next().String()); // host:port                }            }            if (resIsMaster["arbiters"].isABSONObj()) {                BSONObjIterator piter(resIsMaster["arbiters"].Obj());                while (piter.more()) {                    hostSet.insert(piter.next().String()); // host:port                }            }            vector<HostAndPort> hosts = shardConnectionString.getServers();            for (size_t i = 0; i < hosts.size(); i++) {                if (!hosts[i].hasPort()) {                    hosts[i] = HostAndPort(hosts[i].host(), hosts[i].port());                }                string host = hosts[i].toString(); // host:port                if (hostSet.find(host) == hostSet.end()) {                    offendingHost = host;                    foundAll = false;                    break;                }            }        }        if (!foundAll) {            return Status(ErrorCodes::OperationFailed,                          str::stream() << "in seed list " << shardConnectionString.toString()                                        << ", host " << offendingHost                                        << " does not belong to replica set " << setName);//.........这里部分代码省略.........

    Status parseAndValidateCreateUserCommand(const BSONObj& cmdObj,                                             const std::string& dbname,                                             AuthorizationManager* authzManager,                                             BSONObj* parsedUserObj) {        unordered_set<std::string> validFieldNames;        validFieldNames.insert("createUser");        validFieldNames.insert("customData");        validFieldNames.insert("pwd");        validFieldNames.insert("roles");        validFieldNames.insert("writeConcern");        // Iterate through all fields in command object and make sure there are no unexpected        // ones.        for (BSONObjIterator iter(cmdObj); iter.more(); iter.next()) {            StringData fieldName = (*iter).fieldNameStringData();            if (!validFieldNames.count(fieldName.toString())) {                return Status(ErrorCodes::BadValue,                              mongoutils::str::stream() << "/"" << fieldName << "/" is not "                                      "a valid argument to createUser");            }        }        BSONObjBuilder userObjBuilder;        userObjBuilder.append("_id", OID::gen());        // Parse user name        std::string userName;        Status status = bsonExtractStringField(cmdObj, "createUser", &userName);        if (!status.isOK()) {            return status;        }        // Prevent creating users in the local database        if (dbname == "local") {            return Status(ErrorCodes::BadValue, "Cannot create users in the local database");        }        userObjBuilder.append(AuthorizationManager::USER_NAME_FIELD_NAME, userName);        userObjBuilder.append(AuthorizationManager::USER_SOURCE_FIELD_NAME, dbname);        // Parse password        if (cmdObj.hasField("pwd")) {            std::string clearTextPassword;            status = bsonExtractStringField(cmdObj, "pwd", &clearTextPassword);            if (!status.isOK()) {                return status;            }            std::string password = auth::createPasswordDigest(userName, clearTextPassword);            userObjBuilder.append("credentials", BSON("MONGODB-CR" << password));        } else {            if (dbname != "$external") {                return Status(ErrorCodes::BadValue,                              "Must provide a 'pwd' field for all user documents, except those"                                      " with '$external' as the user's source");            }        }        // Parse custom data        if (cmdObj.hasField("customData")) {            BSONElement element;            status = bsonExtractTypedField(cmdObj, "customData", Object, &element);            if (!status.isOK()) {                return status;            }            userObjBuilder.append("customData", element.Obj());        }        // Parse roles        if (cmdObj.hasField("roles")) {            BSONElement rolesElement;            status = bsonExtractTypedField(cmdObj, "roles", Array, &rolesElement);            if (!status.isOK()) {                return status;            }            BSONArray modifiedRolesArray;            status = _validateAndModifyRolesArray(rolesElement,                                                  dbname,                                                  authzManager,                                                  &modifiedRolesArray);            if (!status.isOK()) {                return status;            }            userObjBuilder.append("roles", modifiedRolesArray);        }        *parsedUserObj = userObjBuilder.obj();        // Make sure document to insert is valid        V2PrivilegeDocumentParser parser;        status = parser.checkValidPrivilegeDocument(*parsedUserObj);        if (!status.isOK()) {            return status;        }        return Status::OK();    }

        bool run(OperationContext* txn, const string& dbname , BSONObj& jsobj, int, string& errmsg, BSONObjBuilder& result, bool /*fromRepl*/) {            DBDirectClient db(txn);            BSONElement e = jsobj.firstElement();            string toDeleteNs = dbname + '.' + e.valuestr();            LOG(0) << "CMD: reIndex " << toDeleteNs << endl;            Lock::DBWrite dbXLock(txn->lockState(), dbname);            WriteUnitOfWork wunit(txn->recoveryUnit());            Client::Context ctx(txn, toDeleteNs);            Collection* collection = ctx.db()->getCollection( txn, toDeleteNs );            if ( !collection ) {                errmsg = "ns not found";                return false;            }            BackgroundOperation::assertNoBgOpInProgForNs( toDeleteNs );            std::vector<BSONObj> indexesInProg = stopIndexBuilds(txn, ctx.db(), jsobj);            list<BSONObj> all;            auto_ptr<DBClientCursor> i = db.query( dbname + ".system.indexes" , BSON( "ns" << toDeleteNs ) , 0 , 0 , 0 , QueryOption_SlaveOk );            BSONObjBuilder b;            while ( i->more() ) {                const BSONObj spec = i->next().removeField("v").getOwned();                const BSONObj key = spec.getObjectField("key");                const Status keyStatus = validateKeyPattern(key);                if (!keyStatus.isOK()) {                    errmsg = str::stream()                        << "Cannot rebuild index " << spec << ": " << keyStatus.reason()                        << " For more info see http://dochub.mongodb.org/core/index-validation";                    return false;                }                b.append( BSONObjBuilder::numStr( all.size() ) , spec );                all.push_back( spec );            }            result.appendNumber( "nIndexesWas", collection->getIndexCatalog()->numIndexesTotal() );            Status s = collection->getIndexCatalog()->dropAllIndexes(txn, true);            if ( !s.isOK() ) {                errmsg = "dropIndexes failed";                return appendCommandStatus( result, s );            }            for ( list<BSONObj>::iterator i=all.begin(); i!=all.end(); i++ ) {                BSONObj o = *i;                LOG(1) << "reIndex ns: " << toDeleteNs << " index: " << o << endl;                Status s = collection->getIndexCatalog()->createIndex(txn, o, false);                if ( !s.isOK() )                    return appendCommandStatus( result, s );            }            result.append( "nIndexes" , (int)all.size() );            result.appendArray( "indexes" , b.obj() );            IndexBuilder::restoreIndexes(indexesInProg);            wunit.commit();            return true;        }

TEST( AllElemMatchOp, MatchesElement ) {    BSONObj baseOperanda1 = BSON( "a" << 1 );    auto_ptr<ComparisonMatchExpression> eqa1( new ComparisonMatchExpression() );    ASSERT( eqa1->init( "a", ComparisonMatchExpression::EQ, baseOperanda1[ "a" ] ).isOK() );    BSONObj baseOperandb1 = BSON( "b" << 1 );    auto_ptr<ComparisonMatchExpression> eqb1( new ComparisonMatchExpression() );    ASSERT( eqb1->init( "b", ComparisonMatchExpression::EQ, baseOperandb1[ "b" ] ).isOK() );    auto_ptr<AndMatchExpression> and1( new AndMatchExpression() );    and1->add( eqa1.release() );    and1->add( eqb1.release() );    // and1 = { a : 1, b : 1 }    auto_ptr<ElemMatchObjectMatchExpression> elemMatch1( new ElemMatchObjectMatchExpression() );    elemMatch1->init( "x", and1.release() );    // elemMatch1 = { x : { $elemMatch : { a : 1, b : 1 } } }    BSONObj baseOperanda2 = BSON( "a" << 2 );    auto_ptr<ComparisonMatchExpression> eqa2( new ComparisonMatchExpression() );    ASSERT( eqa2->init( "a", ComparisonMatchExpression::EQ, baseOperanda2[ "a" ] ).isOK() );    BSONObj baseOperandb2 = BSON( "b" << 2 );    auto_ptr<ComparisonMatchExpression> eqb2( new ComparisonMatchExpression() );    ASSERT( eqb2->init( "b", ComparisonMatchExpression::EQ, baseOperandb2[ "b" ] ).isOK() );    auto_ptr<AndMatchExpression> and2( new AndMatchExpression() );    and2->add( eqa2.release() );    and2->add( eqb2.release() );    auto_ptr<ElemMatchObjectMatchExpression> elemMatch2( new ElemMatchObjectMatchExpression() );    elemMatch2->init( "x", and2.release() );    // elemMatch2 = { x : { $elemMatch : { a : 2, b : 2 } } }    AllElemMatchOp op;    op.init( "" );    op.add( elemMatch1.release() );    op.add( elemMatch2.release() );    BSONObj nonArray = BSON( "x" << 4 );    ASSERT( !op.matchesSingleElement( nonArray[ "x" ] ) );    BSONObj emptyArray = BSON( "x" << BSONArray() );    ASSERT( !op.matchesSingleElement( emptyArray[ "x" ] ) );    BSONObj nonObjArray = BSON( "x" << BSON_ARRAY( 4 ) );    ASSERT( !op.matchesSingleElement( nonObjArray[ "x" ] ) );    BSONObj singleObjMatch = BSON( "x" << BSON_ARRAY( BSON( "a" << 1 << "b" << 1 ) ) );    ASSERT( !op.matchesSingleElement( singleObjMatch[ "x" ] ) );    BSONObj otherObjMatch = BSON( "x" << BSON_ARRAY( BSON( "a" << 2 << "b" << 2 ) ) );    ASSERT( !op.matchesSingleElement( otherObjMatch[ "x" ] ) );    BSONObj bothObjMatch = BSON( "x" << BSON_ARRAY( BSON( "a" << 1 << "b" << 1 ) <<                                 BSON( "a" << 2 << "b" << 2 ) ) );    ASSERT( op.matchesSingleElement( bothObjMatch[ "x" ] ) );    BSONObj noObjMatch = BSON( "x" << BSON_ARRAY( BSON( "a" << 1 << "b" << 2 ) <<                               BSON( "a" << 2 << "b" << 1 ) ) );    ASSERT( !op.matchesSingleElement( noObjMatch[ "x" ] ) );}

    void Consensus::_electSelf() {        if( time(0) < steppedDown )            return;        {            const OpTime ord = theReplSet->lastOpTimeWritten;            if( ord == 0 ) {                log() << "replSet info not trying to elect self, do not yet have a complete set of data from any point in time" << rsLog;                return;            }        }        bool allUp;        int nTies;        if( !_weAreFreshest(allUp, nTies) ) {            return;        }        rs.sethbmsg("",9);        if (!allUp && time(0) - serverGlobalParams.started < 60 * 5) {            /* the idea here is that if a bunch of nodes bounce all at once, we don't want to drop data               if we don't have to -- we'd rather be offline and wait a little longer instead               todo: make this configurable.               */            rs.sethbmsg("not electing self, not all members up and we have been up less than 5 minutes");            return;        }        Member& me = *rs._self;        if( nTies ) {            /* tie?  we then randomly sleep to try to not collide on our voting. */            /* todo: smarter. */            if( me.id() == 0 || _sleptLast ) {                // would be fine for one node not to sleep                // todo: biggest / highest priority nodes should be the ones that get to not sleep            }            else {                verify( !rs.lockedByMe() ); // bad to go to sleep locked                unsigned ms = ((unsigned) rand()) % 1000 + 50;                DEV log() << "replSet tie " << nTies << " sleeping a little " << ms << "ms" << rsLog;                _sleptLast = true;                sleepmillis(ms);                throw RetryAfterSleepException();            }        }        _sleptLast = false;        time_t start = time(0);        unsigned meid = me.id();        int tally = _yea( meid );        bool success = false;        try {            log() << "replSet info electSelf " << meid << rsLog;            BSONObj electCmd = BSON(                                   "replSetElect" << 1 <<                                   "set" << rs.name() <<                                   "who" << me.fullName() <<                                   "whoid" << me.hbinfo().id() <<                                   "cfgver" << rs._cfg->version <<                                   "round" << OID::gen() /* this is just for diagnostics */                               );            int configVersion;            list<Target> L;            rs.getTargets(L, configVersion);            _multiCommand(electCmd, L);            {                for( list<Target>::iterator i = L.begin(); i != L.end(); i++ ) {                    LOG(1) << "replSet elect res: " << i->result.toString() << rsLog;                    if( i->ok ) {                        int v = i->result["vote"].Int();                        tally += v;                    }                }                if( tally*2 <= _totalVotes() ) {                    log() << "replSet couldn't elect self, only received " << tally << " votes" << rsLog;                }                else if( time(0) - start > 30 ) {                    // defensive; should never happen as we have timeouts on connection and operation for our conn                    log() << "replSet too much time passed during our election, ignoring result" << rsLog;                }                else if( configVersion != rs.config().version ) {                    log() << "replSet config version changed during our election, ignoring result" << rsLog;                }                else {                    /* succeeded. */                    LOG(1) << "replSet election succeeded, assuming primary role" << rsLog;                    success = true;                    setElectionTime(getNextGlobalOptime());                    rs._assumePrimary();                }            }        }        catch( std::exception& ) {//.........这里部分代码省略.........

    void WriteBackListener::run() {        int secsToSleep = 0;        scoped_ptr<ChunkVersion> lastNeededVersion;        int lastNeededCount = 0;        bool needsToReloadShardInfo = false;        while ( ! inShutdown() ) {            if ( ! Shard::isAShardNode( _addr ) ) {                LOG(1) << _addr << " is not a shard node" << endl;                sleepsecs( 60 );                continue;            }            try {                if (needsToReloadShardInfo) {                    // It's possible this shard was removed                    Shard::reloadShardInfo();                    needsToReloadShardInfo = false;                }                ScopedDbConnection conn(_addr);                BSONObj result;                {                    BSONObjBuilder cmd;                    cmd.appendOID( "writebacklisten" , &serverID ); // Command will block for data                    if ( ! conn->runCommand( "admin" , cmd.obj() , result ) ) {                        result = result.getOwned();                        log() <<  "writebacklisten command failed!  "  << result << endl;                        conn.done();                        continue;                    }                }                conn.done();                LOG(1) << "writebacklisten result: " << result << endl;                BSONObj data = result.getObjectField( "data" );                if ( data.getBoolField( "writeBack" ) ) {                    string ns = data["ns"].valuestrsafe();                    ConnectionIdent cid( "" , 0 );                    OID wid;                    if ( data["connectionId"].isNumber() && data["id"].type() == jstOID ) {                        string s = "";                        if ( data["instanceIdent"].type() == String )                            s = data["instanceIdent"].String();                        cid = ConnectionIdent( s , data["connectionId"].numberLong() );                        wid = data["id"].OID();                    }                    else {                        warning() << "mongos/mongod version mismatch (1.7.5 is the split)" << endl;                    }                    int len; // not used, but needed for next call                    Message msg( (void*)data["msg"].binData( len ) , false );                    massert( 10427 ,  "invalid writeback message" , msg.header()->valid() );                    DBConfigPtr db = grid.getDBConfig( ns );                    ChunkVersion needVersion = ChunkVersion::fromBSON( data, "version" );                    //                    // TODO: Refactor the sharded strategy to correctly handle all sharding state changes itself,                    // we can't rely on WBL to do this for us b/c anything could reset our state in-between.                    // We should always reload here for efficiency when possible, but staleness is also caught in the                    // loop below.                    //                    ChunkManagerPtr manager;                    ShardPtr primary;                    db->getChunkManagerOrPrimary( ns, manager, primary );                    ChunkVersion currVersion;                    if( manager ) currVersion = manager->getVersion();                    LOG(1) << "connectionId: " << cid << " writebackId: " << wid << " needVersion : " << needVersion.toString()                           << " mine : " << currVersion.toString() << endl;                    LOG(1) << msg.toString() << endl;                    //                    // We should reload only if we need to update our version to be compatible *and* we                    // haven't already done so.  This avoids lots of reloading when we remove/add a sharded collection                    //                    bool alreadyReloaded = lastNeededVersion &&                                           lastNeededVersion->isEquivalentTo( needVersion );                    if( alreadyReloaded ){                        LOG(1) << "wbl already reloaded config information for version "                               << needVersion << ", at version " << currVersion << endl;                    }                    else if( lastNeededVersion ) {                        log() << "new version change detected to " << needVersion.toString()//.........这里部分代码省略.........

    void Balancer::_doBalanceRound( DBClientBase& conn, vector<CandidateChunkPtr>* candidateChunks ) {        assert( candidateChunks );        //        // 1. Check whether there is any sharded collection to be balanced by querying        // the ShardsNS::collections collection        //        auto_ptr<DBClientCursor> cursor = conn.query( ShardNS::collection , BSONObj() );        vector< string > collections;        while ( cursor->more() ) {            BSONObj col = cursor->nextSafe();            // sharded collections will have a shard "key".            if ( ! col["key"].eoo() )                collections.push_back( col["_id"].String() );        }        cursor.reset();        if ( collections.empty() ) {            LOG(1) << "no collections to balance" << endl;            return;        }        //        // 2. Get a list of all the shards that are participating in this balance round        // along with any maximum allowed quotas and current utilization. We get the        // latter by issuing db.serverStatus() (mem.mapped) to all shards.        //        // TODO: skip unresponsive shards and mark information as stale.        //        vector<Shard> allShards;        Shard::getAllShards( allShards );        if ( allShards.size() < 2) {            LOG(1) << "can't balance without more active shards" << endl;            return;        }        map< string, BSONObj > shardLimitsMap;        for ( vector<Shard>::const_iterator it = allShards.begin(); it != allShards.end(); ++it ) {            const Shard& s = *it;            ShardStatus status = s.getStatus();            BSONObj limitsObj = BSON( ShardFields::maxSize( s.getMaxSize() ) <<                                      LimitsFields::currSize( status.mapped() ) <<                                      ShardFields::draining( s.isDraining() )  <<                                      LimitsFields::hasOpsQueued( status.hasOpsQueued() )                                    );            shardLimitsMap[ s.getName() ] = limitsObj;        }        //        // 3. For each collection, check if the balancing policy recommends moving anything around.        //        for (vector<string>::const_iterator it = collections.begin(); it != collections.end(); ++it ) {            const string& ns = *it;            map< string,vector<BSONObj> > shardToChunksMap;            cursor = conn.query( ShardNS::chunk , QUERY( "ns" << ns ).sort( "min" ) );            while ( cursor->more() ) {                BSONObj chunk = cursor->nextSafe();                if ( chunk["jumbo"].trueValue() )                    continue;                vector<BSONObj>& chunks = shardToChunksMap[chunk["shard"].String()];                chunks.push_back( chunk.getOwned() );            }            cursor.reset();            if (shardToChunksMap.empty()) {                LOG(1) << "skipping empty collection (" << ns << ")";                continue;            }            for ( vector<Shard>::iterator i=allShards.begin(); i!=allShards.end(); ++i ) {                // this just makes sure there is an entry in shardToChunksMap for every shard                Shard s = *i;                shardToChunksMap[s.getName()].size();            }            CandidateChunk* p = _policy->balance( ns , shardLimitsMap , shardToChunksMap , _balancedLastTime );            if ( p ) candidateChunks->push_back( CandidateChunkPtr( p ) );        }    }

 BSONObj IntervalBtreeCursor::prettyIndexBounds() const {     return BSON( "lower" << _lowerBound.replaceFieldNames( _indexDetails.keyPattern() ) <<                  "upper" << _upperBound.replaceFieldNames( _indexDetails.keyPattern() ) ); }

    Query ConfigDiffTracker<ValType,ShardType>::        configDiffQuery( const set<ChunkVersion>& extraMinorVersions ) const    {        verifyAttached();        //        // Basic idea behind the query is to find all the chunks $gt the current max version, and        // then also update chunks that we need minor versions - splits and (2.0) max chunks on        // shards        //        static const int maxMinorVersionClauses = 50;        BSONObjBuilder queryB;        int numStaleMinorClauses = extraMinorVersions.size() + _maxShardVersions->size();#ifdef _DEBUG        // In debug builds, randomly trigger full reloads to exercise both codepaths        if( rand() % 2 ) numStaleMinorClauses = maxMinorVersionClauses;#endif        queryB.append(ChunkType::ns(), _ns);        //        // If we have only a few minor versions to refresh, we can be more selective in our query        //        if( numStaleMinorClauses < maxMinorVersionClauses ){            //            // Get any version changes higher than we know currently            //            BSONArrayBuilder queryOrB( queryB.subarrayStart( "$or" ) );            {                BSONObjBuilder queryNewB( queryOrB.subobjStart() );                {                    BSONObjBuilder ts(queryNewB.subobjStart(ChunkType::DEPRECATED_lastmod()));                    // We should *always* pull at least a single chunk back, this lets us quickly                    // detect if our collection was unsharded (and most of the time if it was                    // resharded) in the meantime                    ts.appendTimestamp( "$gte", _maxVersion->toLong() );                    ts.done();                }                queryNewB.done();            }            // Get any shard version changes higher than we know currently            // Needed since there could have been a split of the max version chunk of any shard            // TODO: Ideally, we shouldn't care about these            for( typename map<ShardType, ChunkVersion>::const_iterator it = _maxShardVersions->begin(); it != _maxShardVersions->end(); it++ ){                BSONObjBuilder queryShardB( queryOrB.subobjStart() );                queryShardB.append(ChunkType::shard(), nameFrom( it->first ) );                {                    BSONObjBuilder ts(queryShardB.subobjStart(ChunkType::DEPRECATED_lastmod()));                    ts.appendTimestamp( "$gt", it->second.toLong() );                    ts.done();                }                queryShardB.done();            }            // Get any minor version changes we've marked as interesting            // TODO: Ideally we shouldn't care about these            for( set<ChunkVersion>::const_iterator it = extraMinorVersions.begin(); it != extraMinorVersions.end(); it++ ){                BSONObjBuilder queryShardB( queryOrB.subobjStart() );                {                    BSONObjBuilder ts(queryShardB.subobjStart(ChunkType::DEPRECATED_lastmod()));                    ts.appendTimestamp( "$gt", it->toLong() );                    ts.appendTimestamp( "$lt",                                        ChunkVersion( it->majorVersion() + 1, 0, OID() ).toLong() );                    ts.done();                }                queryShardB.done();            }            queryOrB.done();        }        BSONObj query = queryB.obj();        LOG(2) << "major version query from " << *_maxVersion << " and over "               << _maxShardVersions->size() << " shards is " << query << endl;        //        // NOTE: IT IS IMPORTANT FOR CONSISTENCY THAT WE SORT BY ASC VERSION, TO HANDLE        // CURSOR YIELDING BETWEEN CHUNKS BEING MIGRATED.        //        // This ensures that changes to chunk version (which will always be higher) will always        // come *after* our current position in the chunk cursor.        //        Query queryObj(query);        queryObj.sort(BSON( "lastmod" << 1 ));        return Query( query );    }

        bool handleSpecialNamespaces( Request& r , QueryMessage& q ) {            const char * ns = r.getns();            ns = strstr( r.getns() , ".$cmd.sys." );            if ( ! ns )                return false;            ns += 10;            BSONObjBuilder b;            vector<Shard> shards;            if ( strcmp( ns , "inprog" ) == 0 ) {                Shard::getAllShards( shards );                BSONArrayBuilder arr( b.subarrayStart( "inprog" ) );                for ( unsigned i=0; i<shards.size(); i++ ) {                    Shard shard = shards[i];                    ScopedDbConnection conn( shard );                    BSONObj temp = conn->findOne( r.getns() , BSONObj() );                    if ( temp["inprog"].isABSONObj() ) {                        BSONObjIterator i( temp["inprog"].Obj() );                        while ( i.more() ) {                            BSONObjBuilder x;                            BSONObjIterator j( i.next().Obj() );                            while( j.more() ) {                                BSONElement e = j.next();                                if ( str::equals( e.fieldName() , "opid" ) ) {                                    stringstream ss;                                    ss << shard.getName() << ':' << e.numberInt();                                    x.append( "opid" , ss.str() );                                }                                else if ( str::equals( e.fieldName() , "client" ) ) {                                    x.appendAs( e , "client_s" );                                }                                else {                                    x.append( e );                                }                            }                            arr.append( x.obj() );                        }                    }                    conn.done();                }                arr.done();            }            else if ( strcmp( ns , "killop" ) == 0 ) {                BSONElement e = q.query["op"];                if ( strstr( r.getns() , "admin." ) != 0 ) {                    b.append( "err" , "unauthorized" );                }                else if ( e.type() != String ) {                    b.append( "err" , "bad op" );                    b.append( e );                }                else {                    b.append( e );                    string s = e.String();                    string::size_type i = s.find( ':' );                    if ( i == string::npos ) {                        b.append( "err" , "bad opid" );                    }                    else {                        string shard = s.substr( 0 , i );                        int opid = atoi( s.substr( i + 1 ).c_str() );                        b.append( "shard" , shard );                        b.append( "shardid" , opid );                        log() << "want to kill op: " << e << endl;                        Shard s(shard);                        ScopedDbConnection conn( s );                        conn->findOne( r.getns() , BSON( "op" << opid ) );                        conn.done();                    }                }            }            else if ( strcmp( ns , "unlock" ) == 0 ) {                b.append( "err" , "can't do unlock through mongos" );            }            else {                log( LL_WARNING ) << "unknown sys command [" << ns << "]" << endl;                return false;            }            BSONObj x = b.done();            replyToQuery(0, r.p(), r.m(), x);            return true;        }

BSONObj RawMongoProgramOutput(const BSONObj& args, void* data) {    return BSON("" << programOutputLogger.str());}

TEST_F(SyncTailTest, SyncApplyNoNamespaceBadOp) {    const BSONObj op = BSON("op"                            << "x");    ASSERT_OK(SyncTail::syncApply(_txn.get(), op, false, _applyOp, _applyCmd, _incOps));    ASSERT_EQUALS(0U, _opsApplied);}

    Status parseAndValidateUpdateUserCommand(const BSONObj& cmdObj,                                             const std::string& dbname,                                             AuthorizationManager* authzManager,                                             BSONObj* parsedUpdateObj,                                             UserName* parsedUserName) {        unordered_set<std::string> validFieldNames;        validFieldNames.insert("updateUser");        validFieldNames.insert("customData");        validFieldNames.insert("pwd");        validFieldNames.insert("roles");        validFieldNames.insert("writeConcern");        // Iterate through all fields in command object and make sure there are no unexpected        // ones.        for (BSONObjIterator iter(cmdObj); iter.more(); iter.next()) {            StringData fieldName = (*iter).fieldNameStringData();            if (!validFieldNames.count(fieldName.toString())) {                return Status(ErrorCodes::BadValue,                              mongoutils::str::stream() << "/"" << fieldName << "/" is not "                                      "a valid argument to createUser");            }        }        BSONObjBuilder updateSetBuilder;        // Parse user name        std::string userName;        Status status = bsonExtractStringField(cmdObj, "updateUser", &userName);        if (!status.isOK()) {            return status;        }        *parsedUserName = UserName(userName, dbname);        // Parse password        if (cmdObj.hasField("pwd")) {            std::string clearTextPassword;            status = bsonExtractStringField(cmdObj, "pwd", &clearTextPassword);            if (!status.isOK()) {                return status;            }            std::string password = auth::createPasswordDigest(userName, clearTextPassword);            updateSetBuilder.append("credentials.MONGODB-CR", password);        }        // Parse custom data        if (cmdObj.hasField("customData")) {            BSONElement element;            status = bsonExtractTypedField(cmdObj, "customData", Object, &element);            if (!status.isOK()) {                return status;            }            updateSetBuilder.append("customData", element.Obj());        }        // Parse roles        if (cmdObj.hasField("roles")) {            BSONElement rolesElement;            Status status = bsonExtractTypedField(cmdObj, "roles", Array, &rolesElement);            if (!status.isOK()) {                return status;            }            BSONArray modifiedRolesObj;            status = _validateAndModifyRolesArray(rolesElement,                                                  dbname,                                                  authzManager,                                                  &modifiedRolesObj);            if (!status.isOK()) {                return status;            }            updateSetBuilder.append("roles", modifiedRolesObj);        }        BSONObj updateSet = updateSetBuilder.obj();        if (updateSet.isEmpty()) {            return Status(ErrorCodes::UserModificationFailed,                          "Must specify at least one field to update in updateUser");        }        *parsedUpdateObj = BSON("$set" << updateSet);        return Status::OK();    }

 void ModMatchExpression::toBSON(BSONObjBuilder* out) const {     out->append(path(), BSON("$mod" << BSON_ARRAY(_divisor << _remainder))); }

 Base() {     db.dropCollection(ns());     db.ensureIndex(ns(), BSON( "files_id" << 1 << "n" << 1 )); }

 DevNullRecordStore( StringData ns, const CollectionOptions& options )     : RecordStore( ns ), _options( options ) {     _numInserts = 0;     _dummy = BSON( "_id" << 1 ); }

 void run() {     IndexDescriptor* id = addIndexWithInfo();     // Create a SortPhaseOne.     SortPhaseOne phaseOne;     phaseOne.sorter.reset(new BSONObjExternalSorter(_aFirstSort));     // It's necessary to index sufficient keys that a RARELY condition will be triggered,     // but few enough keys that the btree builder will not create an internal node and check     // for an interrupt internally (which would cause this test to pass spuriously).     int32_t nKeys = 130;     // Add index keys to the phaseOne.     for( int32_t i = 0; i < nKeys; ++i ) {         phaseOne.sorter->add( BSON( "a" << i ), /* dummy disk loc */ DiskLoc(), false );     }     phaseOne.nkeys = phaseOne.n = nKeys;     phaseOne.sorter->sort( false );     // Set up remaining arguments.     set<DiskLoc> dups;     CurOp* op = cc().curop();     ProgressMeterHolder pm (op->setMessage("InterruptBuildBottomUp",                                            "InterruptBuildBottomUp Progress",                                            nKeys,                                            nKeys));     pm.finished();     Timer timer;     // The index's root has not yet been set.     ASSERT( id->getHead().isNull() );     // Register a request to kill the current operation.     cc().curop()->kill();     if ( _mayInterrupt ) {         // The build is aborted due to the kill request.         ASSERT_THROWS                 ( buildBottomUpPhases2And3<V1>( true,                                                 id,                                                 *phaseOne.sorter,                                                 false,                                                 dups,                                                 op,                                                 &phaseOne,                                                 pm,                                                 timer,                                                 _mayInterrupt ),                   UserException );         // The root of the index is not set because the build did not complete.         ASSERT( id->getHead().isNull() );     }     else {         // The build is aborted despite the kill request because mayInterrupt == false.         buildBottomUpPhases2And3<V1>( true,                                       id,                                       *phaseOne.sorter,                                       false,                                       dups,                                       op,                                       &phaseOne,                                       pm,                                       timer,                                       _mayInterrupt );         // The index's root is set after the build is complete.         ASSERT( !id->getHead().isNull() );     } }

    bool DBConfig::dropDatabase( string& errmsg ) {        /**         * 1) make sure everything is up         * 2) update config server         * 3) drop and reset sharded collections         * 4) drop and reset primary         * 5) drop everywhere to clean up loose ends         */        log() << "DBConfig::dropDatabase: " << _name << endl;        configServer.logChange( "dropDatabase.start" , _name , BSONObj() );        // 1        if ( ! configServer.allUp( errmsg ) ) {            log(1) << "/t DBConfig::dropDatabase not all up" << endl;            return 0;        }        // 2        grid.removeDB( _name );        {            ScopedDbConnection conn( configServer.modelServer() );            conn->remove( ShardNS::database , BSON( "_id" << _name ) );            errmsg = conn->getLastError();            if ( ! errmsg.empty() ) {                log() << "could not drop '" << _name << "': " << errmsg << endl;                conn.done();                return false;            }            conn.done();        }        if ( ! configServer.allUp( errmsg ) ) {            log() << "error removing from config server even after checking!" << endl;            return 0;        }        log(1) << "/t removed entry from config server for: " << _name << endl;        set<Shard> allServers;        // 3        while ( true ) {            int num = 0;            if ( ! _dropShardedCollections( num , allServers , errmsg ) )                return 0;            log() << "   DBConfig::dropDatabase: " << _name << " dropped sharded collections: " << num << endl;            if ( num == 0 )                break;        }        // 4        {            ScopedDbConnection conn( _primary );            BSONObj res;            if ( ! conn->dropDatabase( _name , &res ) ) {                errmsg = res.toString();                return 0;            }            conn.done();        }        // 5        for ( set<Shard>::iterator i=allServers.begin(); i!=allServers.end(); i++ ) {            ScopedDbConnection conn( *i );            BSONObj res;            if ( ! conn->dropDatabase( _name , &res ) ) {                errmsg = res.toString();                return 0;            }            conn.done();        }        log(1) << "/t dropped primary db for: " << _name << endl;        configServer.logChange( "dropDatabase" , _name , BSONObj() );        return true;    }

 SimpleCompoundIndex() {     _client.insert("unittests.system.indexes",             BSON("name" << "x"                  << "ns" << _ns                  << "key" << BSON("x" << 1 << "y" << 1))); }

 void addUniqueIndex() {     addOp("i", BSON("ns" << ns() << "key" << BSON("x" << 1) << "name" << "x1" << "unique" << true), 0, "unittests.system.indexes");     addInserts(2); }