dbclientinterface.h

Go to the documentation of this file.

 
/*    Copyright 2009 10gen Inc.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
 
#pragma once
 
#include <boost/thread/locks.hpp>
#include <boost/thread/mutex.hpp>
 
#include "mongo/config.h"
 
#include "mongo/base/string_data.h"
#include "mongo/client/bulk_operation_builder.h"
#include "mongo/client/exceptions.h"
#include "mongo/client/export_macros.h"
#include "mongo/client/index_spec.h"
#include "mongo/client/write_concern.h"
#include "mongo/client/write_options.h"
#include "mongo/db/jsobj.h"
#include "mongo/logger/log_severity.h"
#include "mongo/platform/atomic_word.h"
#include "mongo/stdx/functional.h"
#include "mongo/util/mongoutils/str.h"
#include "mongo/util/net/message.h"
#include "mongo/util/net/message_port.h"
 
namespace mongo {
 
enum MONGO_CLIENT_API QueryOptions {
    QueryOption_CursorTailable = 1 << 1,
 
    QueryOption_SlaveOk = 1 << 2,
 
    // findingStart mode is used to find the first operation of interest when
    // we are scanning through a repl log.  For efficiency in the common case,
    // where the first operation of interest is closer to the tail than the head,
    // we start from the tail of the log and work backwards until we find the
    // first operation of interest.  Then we scan forward from that first operation,
    // actually returning results to the client.  During the findingStart phase,
    // we release the db mutex occasionally to avoid blocking the db process for
    // an extended period of time.
    QueryOption_OplogReplay = 1 << 3,
 
    QueryOption_NoCursorTimeout = 1 << 4,
 
    QueryOption_AwaitData = 1 << 5,
 
    QueryOption_Exhaust = 1 << 6,
 
    QueryOption_PartialResults = 1 << 7,
 
    QueryOption_AllSupported = QueryOption_CursorTailable | QueryOption_SlaveOk |
        QueryOption_OplogReplay | QueryOption_NoCursorTimeout | QueryOption_AwaitData |
        QueryOption_Exhaust | QueryOption_PartialResults,
};
 
enum MONGO_CLIENT_API WriteOptions {
    WriteOption_FromWriteback = 1 << 31
};
 
//
// For legacy reasons, the reserved field pre-namespace of certain types of messages is used
// to store options as opposed to the flags after the namespace.  This should be transparent to
// the api user, but we need these constants to disassemble/reassemble the messages correctly.
//
 
enum MONGO_CLIENT_API ReservedOptions {
    Reserved_InsertOption_ContinueOnError = 1 << 0,
    Reserved_FromWriteback = 1 << 1
};
 
enum MONGO_CLIENT_API ReadPreference {
    ReadPreference_PrimaryOnly = 0,
 
    ReadPreference_PrimaryPreferred,
 
    ReadPreference_SecondaryOnly,
 
    ReadPreference_SecondaryPreferred,
 
    ReadPreference_Nearest,
};
 
class MONGO_CLIENT_API DBClientBase;
class MONGO_CLIENT_API DBClientConnection;
 
class MONGO_CLIENT_API ConnectionString {
public:
    enum ConnectionType { INVALID, MASTER, PAIR, SET, CUSTOM };
 
    ConnectionString() {
        _type = INVALID;
    }
 
    // Note: This should only be used for direct connections to a single server.  For replica
    // set and SyncClusterConnections, use ConnectionString::parse.
    ConnectionString(const HostAndPort& server) {
        _type = MASTER;
        _servers.push_back(server);
        _finishInit();
    }
 
    ConnectionString(ConnectionType type, const std::string& s, const std::string& setName = "") {
        _type = type;
        _setName = setName;
        _fillServers(s);
 
        switch (_type) {
            case MASTER:
                verify(_servers.size() == 1);
                break;
            case SET:
                verify(_setName.size());
                verify(_servers.size() >= 1);  // 1 is ok since we can derive
                break;
            case PAIR:
                verify(_servers.size() == 2);
                break;
            default:
                verify(_servers.size() > 0);
        }
 
        _finishInit();
    }
 
    ConnectionString(const std::string& s, ConnectionType favoredMultipleType) {
        _type = INVALID;
 
        _fillServers(s);
        if (_type != INVALID) {
            // set already
        } else if (_servers.size() == 1) {
            _type = MASTER;
        } else {
            _type = favoredMultipleType;
            verify(_type == SET);
        }
        _finishInit();
    }
 
    bool isValid() const {
        return _type != INVALID;
    }
 
    std::string toString() const {
        return _string;
    }
 
    DBClientBase* connect(std::string& errmsg, double socketTimeout = 0) const;
 
    std::string getSetName() const {
        return _setName;
    }
 
    const std::vector<HostAndPort>& getServers() const {
        return _servers;
    }
 
    ConnectionType type() const {
        return _type;
    }
 
    const std::string& getUser() const {
        return _user;
    }
 
    const std::string& getPassword() const {
        return _password;
    }
 
    const BSONObj& getOptions() const {
        return _options;
    }
 
    const std::string& getDatabase() const {
        return _database;
    }
 
    bool sameLogicalEndpoint(const ConnectionString& other) const;
 
    static ConnectionString MONGO_CLIENT_FUNC
    parse(const std::string& address, std::string& errmsg);
 
    static ConnectionString MONGO_CLIENT_FUNC
    parseDeprecated(const std::string& address, std::string& errmsg);
 
    static std::string MONGO_CLIENT_FUNC typeToString(ConnectionType type);
 
    //
    // Allow overriding the default connection behavior
    // This is needed for some tests, which otherwise would fail because they are unable to contact
    // the correct servers.
    //
 
    class ConnectionHook {
    public:
        virtual ~ConnectionHook() {}
 
        // Returns an alternative connection object for a string
        virtual DBClientBase* connect(const ConnectionString& c,
                                      std::string& errmsg,
                                      double socketTimeout) = 0;
    };
 
    static void setConnectionHook(ConnectionHook* hook) {
        boost::lock_guard<boost::mutex> lk(_connectHookMutex);
        _connectHook = hook;
    }
 
    static ConnectionHook* getConnectionHook() {
        boost::lock_guard<boost::mutex> lk(_connectHookMutex);
        return _connectHook;
    }
 
    // Allows ConnectionStrings to be stored more easily in sets/maps
    bool operator<(const ConnectionString& other) const {
        return _string < other._string;
    }
 
    //
    // FOR TESTING ONLY - useful to be able to directly mock a connection string without
    // including the entire client library.
    //
 
    static ConnectionString mock(const HostAndPort& server) {
        ConnectionString connStr;
        connStr._servers.push_back(server);
        connStr._string = server.toString();
        return connStr;
    }
 
private:
    ConnectionString(ConnectionType type,
                     const std::string& user,
                     const std::string& password,
                     const std::string& servers,
                     const std::string& database,
                     const std::string& setName,
                     const BSONObj& options)
        : _type(type),
          _servers(),
          _setName(setName),
          _user(user),
          _password(password),
          _database(database),
          _options(options) {
        _fillServers(servers, false);
        switch (_type) {
            case MASTER:
                verify(_servers.size() == 1);
                break;
            case SET:
                verify(_setName.size());
                verify(_servers.size() >= 1);  // 1 is ok since we can derive
                break;
            case PAIR:
                verify(_servers.size() == 2);
                break;
            default:
                verify(_servers.size() > 0);
        }
 
        _finishInit();
    }
 
    static ConnectionString _parseURL(const std::string& url, std::string& errmsg);
 
    void _fillServers(std::string s, bool legacy = true);
    void _finishInit();
 
    BSONObj _makeAuthObjFromOptions(int maxWireVersion) const;
 
    ConnectionType _type;
    std::vector<HostAndPort> _servers;
    std::string _string;
    std::string _setName;
 
    std::string _user;
    std::string _password;
 
    std::string _database;
    BSONObj _options;
 
    static boost::mutex _connectHookMutex;
    static ConnectionHook* _connectHook;
};
 
class BSONObj;
class DBClientCursor;
class DBClientCursorBatchIterator;
 
class MONGO_CLIENT_API Query {
public:
    static const BSONField<BSONObj> ReadPrefField;
    static const BSONField<std::string> ReadPrefModeField;
    static const BSONField<BSONArray> ReadPrefTagsField;
 
    BSONObj obj;
    Query() : obj(BSONObj()) {}
    Query(const BSONObj& b) : obj(b) {}
    Query(const std::string& json);
    Query(const char* json);
 
    Query& sort(const BSONObj& sortPattern);
 
    Query& sort(const std::string& field, int asc = 1) {
        sort(BSON(field << asc));
        return *this;
    }
 
    Query& hint(BSONObj keyPattern);
    Query& hint(const std::string& indexName);
 
    Query& maxTimeMs(int millis);
 
    Query& minKey(const BSONObj& val);
    Query& maxKey(const BSONObj& val);
 
    Query& explain();
 
    Query& snapshot();
 
    Query& where(const std::string& jscode, BSONObj scope);
    Query& where(const std::string& jscode) {
        return where(jscode, BSONObj());
    }
 
    Query& readPref(ReadPreference pref, const BSONArray& tags);
 
    bool isComplex(bool* hasDollar = 0) const;
    BSONObj getModifiers() const;
    static bool MONGO_CLIENT_FUNC isComplex(const BSONObj& obj, bool* hasDollar = 0);
 
    BSONObj getFilter() const;
    BSONObj getSort() const;
    BSONElement getHint() const;
    BSONObj getReadPref() const;
    int getMaxTimeMs() const;
    bool isExplain() const;
 
    static bool MONGO_CLIENT_FUNC hasReadPreference(const BSONObj& queryObj);
    bool hasReadPreference() const;
    bool hasHint() const;
    bool hasMaxTimeMs() const;
 
    std::string toString() const;
    operator std::string() const {
        return toString();
    }
 
private:
    void makeComplex();
    template <class T>
    void appendComplex(const char* fieldName, const T& val) {
        makeComplex();
        BSONObjBuilder b;
        b.appendElements(obj);
        b.append(fieldName, val);
        obj = b.obj();
    }
};
 
class MONGO_CLIENT_API QuerySpec {
    std::string _ns;
    int _ntoskip;
    int _ntoreturn;
    int _options;
    BSONObj _query;
    BSONObj _fields;
    Query _queryObj;
 
public:
    QuerySpec(const std::string& ns,
              const BSONObj& query,
              const BSONObj& fields,
              int ntoskip,
              int ntoreturn,
              int options)
        : _ns(ns),
          _ntoskip(ntoskip),
          _ntoreturn(ntoreturn),
          _options(options),
          _query(query.getOwned()),
          _fields(fields.getOwned()),
          _queryObj(_query) {}
 
    QuerySpec() {}
 
    bool isEmpty() const {
        return _ns.size() == 0;
    }
 
    bool isExplain() const {
        return _queryObj.isExplain();
    }
    BSONObj filter() const {
        return _queryObj.getFilter();
    }
 
    BSONElement hint() const {
        return _queryObj.getHint();
    }
    BSONObj sort() const {
        return _queryObj.getSort();
    }
    BSONObj query() const {
        return _query;
    }
    BSONObj fields() const {
        return _fields;
    }
    BSONObj* fieldsData() {
        return &_fields;
    }
 
    // don't love this, but needed downstrem
    const BSONObj* fieldsPtr() const {
        return &_fields;
    }
 
    std::string ns() const {
        return _ns;
    }
    int ntoskip() const {
        return _ntoskip;
    }
    int ntoreturn() const {
        return _ntoreturn;
    }
    int options() const {
        return _options;
    }
 
    void setFields(BSONObj& o) {
        _fields = o.getOwned();
    }
 
    std::string toString() const {
        return str::stream() << "QSpec " << BSON("ns" << _ns << "n2skip" << _ntoskip << "n2return"
                                                      << _ntoreturn << "options" << _options
                                                      << "query" << _query << "fields" << _fields);
    }
};
 
 
#define MONGO_QUERY(x) ::mongo::Query(BSON(x))
 
// Useful utilities for namespaces
MONGO_CLIENT_API std::string MONGO_CLIENT_FUNC nsGetDB(const std::string& ns);
 
MONGO_CLIENT_API std::string MONGO_CLIENT_FUNC nsGetCollection(const std::string& ns);
 
class MONGO_CLIENT_API DBConnector {
public:
    virtual ~DBConnector() {}
    virtual bool call(Message& toSend,
                      Message& response,
                      bool assertOk = true,
                      std::string* actualServer = 0) = 0;
    virtual void say(Message& toSend, bool isRetry = false, std::string* actualServer = 0) = 0;
    virtual void sayPiggyBack(Message& toSend) = 0;
    /* used by QueryOption_Exhaust.  To use that your subclass must implement this. */
    virtual bool recv(Message& m) {
        verify(false);
        return false;
    }
    // In general, for lazy queries, we'll need to say, recv, then checkResponse
    virtual void checkResponse(const char* data,
                               int nReturned,
                               bool* retry = NULL,
                               std::string* targetHost = NULL) {
        if (retry)
            *retry = false;
        if (targetHost)
            *targetHost = "";
    }
    virtual bool lazySupported() const = 0;
};
 
class MONGO_CLIENT_API DBClientInterface : boost::noncopyable {
public:
    virtual std::auto_ptr<DBClientCursor> query(const std::string& ns,
                                                Query query,
                                                int nToReturn = 0,
                                                int nToSkip = 0,
                                                const BSONObj* fieldsToReturn = 0,
                                                int queryOptions = 0,
                                                int batchSize = 0) = 0;
 
    virtual void insert(const std::string& ns,
                        BSONObj obj,
                        int flags = 0,
                        const WriteConcern* wc = NULL) = 0;
 
    virtual void insert(const std::string& ns,
                        const std::vector<BSONObj>& v,
                        int flags = 0,
                        const WriteConcern* wc = NULL) = 0;
 
    virtual void remove(const std::string& ns,
                        Query query,
                        bool justOne = false,
                        const WriteConcern* wc = NULL) = 0;
 
    virtual void remove(const std::string& ns,
                        Query query,
                        int flags,
                        const WriteConcern* wc = NULL) = 0;
 
    virtual void update(const std::string& ns,
                        Query query,
                        BSONObj obj,
                        bool upsert = false,
                        bool multi = false,
                        const WriteConcern* wc = NULL) = 0;
 
    virtual void update(const std::string& ns,
                        Query query,
                        BSONObj obj,
                        int flags,
                        const WriteConcern* wc = NULL) = 0;
 
    virtual ~DBClientInterface() {}
 
    virtual BSONObj findOne(const std::string& ns,
                            const Query& query,
                            const BSONObj* fieldsToReturn = 0,
                            int queryOptions = 0);
 
    void findN(std::vector<BSONObj>& out,
               const std::string& ns,
               Query query,
               int nToReturn,
               int nToSkip = 0,
               const BSONObj* fieldsToReturn = 0,
               int queryOptions = 0,
               int batchSize = 0);
 
    void save(const StringData& ns, const BSONObj& toSave, const WriteConcern* wc = NULL);
 
    virtual std::string getServerAddress() const = 0;
 
    virtual std::auto_ptr<DBClientCursor> getMore(const std::string& ns,
                                                  long long cursorId,
                                                  int nToReturn = 0,
                                                  int options = 0) = 0;
};
 
class MONGO_CLIENT_API DBClientWithCommands : public DBClientInterface {
public:
    logger::LogSeverity _logLevel;
 
    DBClientWithCommands()
        : _logLevel(logger::LogSeverity::Log()),
          _minWireVersion(0),
          _maxWireVersion(0),
          _cachedAvailableOptions((enum QueryOptions)0),
          _haveCachedAvailableOptions(false) {}
 
    bool simpleCommand(const std::string& dbname, BSONObj* info, const std::string& command);
 
    virtual bool runCommand(const std::string& dbname,
                            const BSONObj& cmd,
                            BSONObj& info,
                            int options = 0);
 
    void auth(const BSONObj& params);
 
    bool auth(const std::string& dbname,
              const std::string& username,
              const std::string& pwd,
              std::string& errmsg,
              bool digestPassword = true);
 
    virtual void logout(const std::string& dbname, BSONObj& info);
 
    virtual unsigned long long count(const std::string& ns,
                                     const Query& query = Query(),
                                     int options = 0,
                                     int limit = 0,
                                     int skip = 0);
 
    static std::string MONGO_CLIENT_FUNC
    createPasswordDigest(const std::string& username, const std::string& clearTextPassword);
 
    virtual bool isMaster(bool& isMaster, BSONObj* info = 0);
 
    bool createCollection(const std::string& ns,
                          long long size = 0,
                          bool capped = false,
                          int max = 0,
                          BSONObj* info = 0);
 
    bool createCollectionWithOptions(const std::string& ns,
                                     long long size = 0,
                                     bool capped = false,
                                     int max = 0,
                                     const BSONObj& extraOptions = BSONObj(),
                                     BSONObj* info = 0);
 
    std::string getLastError(
        const std::string& db, bool fsync = false, bool j = false, int w = 0, int wtimeout = 0);
    std::string getLastError(bool fsync = false, bool j = false, int w = 0, int wtimeout = 0);
 
    virtual BSONObj getLastErrorDetailed(
        const std::string& db, bool fsync = false, bool j = false, int w = 0, int wtimeout = 0);
    virtual BSONObj getLastErrorDetailed(bool fsync = false,
                                         bool j = false,
                                         int w = 0,
                                         int wtimeout = 0);
 
    static std::string MONGO_CLIENT_FUNC getLastErrorString(const BSONObj& res);
 
    BSONObj getPrevError();
 
    bool resetError() {
        return simpleCommand("admin", 0, "reseterror");
    }
 
    virtual bool dropCollection(const std::string& ns, BSONObj* info = NULL) {
        std::string db = nsGetDB(ns);
        std::string coll = nsGetCollection(ns);
        uassert(10011, "no collection name", coll.size());
 
        BSONObj temp;
        if (info == NULL) {
            info = &temp;
        }
 
        bool res = runCommand(db.c_str(), BSON("drop" << coll), *info);
        return res;
    }
 
    bool repairDatabase(const std::string& dbname, BSONObj* info = 0) {
        return simpleCommand(dbname, info, "repairDatabase");
    }
 
    bool copyDatabase(const std::string& fromdb,
                      const std::string& todb,
                      const std::string& fromhost = "",
                      const std::string& mechanism = "DEFAULT",
                      const std::string& username = "",
                      const std::string& password = "",
                      BSONObj* info = 0);
 
    enum ProfilingLevel {
        ProfileOff = 0,
        ProfileSlow = 1,  // log very slow (>100ms) operations
        ProfileAll = 2
 
    };
    bool setDbProfilingLevel(const std::string& dbname, ProfilingLevel level, BSONObj* info = 0);
    bool getDbProfilingLevel(const std::string& dbname, ProfilingLevel& level, BSONObj* info = 0);
 
 
    struct MROutput {
        MROutput(const char* collection) : out(BSON("replace" << collection)) {}
        MROutput(const std::string& collection) : out(BSON("replace" << collection)) {}
        MROutput(const BSONObj& obj) : out(obj) {}
 
        BSONObj out;
    };
    static MROutput MRInline;
 
    BSONObj mapreduce(const std::string& ns,
                      const std::string& jsmapf,
                      const std::string& jsreducef,
                      Query query = Query(),
                      MROutput output = MRInline);
 
    void group(const StringData& ns,
               const StringData& jsreduce,
               std::vector<BSONObj>* output,
               const BSONObj& initial = BSONObj(),
               const Query& query = Query(),
               const BSONObj& key = BSONObj(),
               const StringData& finalize = "");
 
    void groupWithKeyFunction(const StringData& ns,
                              const StringData& jsreduce,
                              std::vector<BSONObj>* output,
                              const BSONObj& initial = BSONObj(),
                              const Query& query = Query(),
                              const StringData& jskey = "",
                              const StringData& finalize = "");
 
    BSONObj distinct(const StringData& ns, const StringData& field, const Query& query = Query());
 
    MONGO_CLIENT_DEPRECATED("deprecated in MongoDB 3.0")
    bool eval(const std::string& dbname,
              const std::string& jscode,
              BSONObj& info,
              BSONElement& retValue,
              BSONObj* args = 0,
              bool nolock = false) {
        return evalDeprecated(dbname, jscode, info, retValue, args, nolock);
    }
 
    bool evalDeprecated(const std::string& dbname,
                        const std::string& jscode,
                        BSONObj& info,
                        BSONElement& retValue,
                        BSONObj* args = 0,
                        bool nolock = false);
 
    bool validate(const std::string& ns, bool scandata = true) {
        BSONObj cmd = BSON("validate" << nsGetCollection(ns) << "scandata" << scandata);
        BSONObj info;
        return runCommand(nsGetDB(ns).c_str(), cmd, info);
    }
 
    /* The following helpers are simply more convenient forms of eval() for certain common cases */
 
    /* invocation with no return value of interest -- with or without one simple parameter */
 
    MONGO_CLIENT_DEPRECATED("deprecated in MongoDB 3.0")
    bool eval(const std::string& dbname, const std::string& jscode) {
        return evalDeprecated(dbname, jscode);
    }
 
    bool evalDeprecated(const std::string& dbname, const std::string& jscode);
 
    template <class T>
    MONGO_CLIENT_DEPRECATED("deprecated in MongoDB 3.0") bool eval(const std::string& dbname,
                                                                   const std::string& jscode,
                                                                   T parm1) {
        return evalDeprecated(dbname, jscode, parm1);
    }
 
    template <class T>
    bool evalDeprecated(const std::string& dbname, const std::string& jscode, T parm1) {
        BSONObj info;
        BSONElement retValue;
        BSONObjBuilder b;
        b.append("0", parm1);
        BSONObj args = b.done();
        return eval(dbname, jscode, info, retValue, &args);
    }
 
    template <class T, class NumType>
    MONGO_CLIENT_DEPRECATED("deprecated in MongoDB 3.0") bool eval(const std::string& dbname,
                                                                   const std::string& jscode,
                                                                   T parm1,
                                                                   NumType& ret) {
        return evalDeprecated(dbname, jscode, parm1, ret);
    }
 
    template <class T, class NumType>
    bool evalDeprecated(const std::string& dbname,
                        const std::string& jscode,
                        T parm1,
                        NumType& ret) {
        BSONObj info;
        BSONElement retValue;
        BSONObjBuilder b;
        b.append("0", parm1);
        BSONObj args = b.done();
        if (!eval(dbname, jscode, info, retValue, &args))
            return false;
        ret = (NumType)retValue.number();
        return true;
    }
 
    std::list<std::string> getDatabaseNames();
 
    std::list<std::string> getCollectionNames(const std::string& db,
                                              const BSONObj& filter = BSONObj());
    std::list<BSONObj> getCollectionInfos(const std::string& db, const BSONObj& filter = BSONObj());
 
    std::auto_ptr<DBClientCursor> enumerateCollections(const std::string& db,
                                                       const BSONObj& filter = BSONObj(),
                                                       int batchSize = 0);
 
    bool exists(const std::string& ns);
 
    void createIndex(const StringData& ns, const BSONObj& keys) {
        return createIndex(ns, IndexSpec().addKeys(keys));
    }
 
    virtual void createIndex(const StringData& ns, const IndexSpec& descriptor);
 
    virtual std::list<BSONObj> getIndexSpecs(const std::string& ns, int options = 0);
 
    virtual std::list<std::string> getIndexNames(const std::string& ns, int options = 0);
 
    virtual std::auto_ptr<DBClientCursor> enumerateIndexes(const std::string& ns,
                                                           int options = 0,
                                                           int batchSize = 0);
 
    virtual void dropIndex(const std::string& ns, BSONObj keys);
    virtual void dropIndex(const std::string& ns, const std::string& indexName);
 
    virtual void dropIndexes(const std::string& ns);
 
    virtual void reIndex(const std::string& ns);
 
    static std::string genIndexName(const BSONObj& keys);
 
    virtual bool dropDatabase(const std::string& dbname, BSONObj* info = 0) {
        bool ret = simpleCommand(dbname, info, "dropDatabase");
        return ret;
    }
 
    virtual std::string toString() const = 0;
 
    typedef stdx::function<void(BSONObjBuilder*)> RunCommandHookFunc;
    virtual void setRunCommandHook(RunCommandHookFunc func);
    RunCommandHookFunc getRunCommandHook() const {
        return _runCommandHook;
    }
 
    typedef stdx::function<void(const BSONObj&, const std::string&)> PostRunCommandHookFunc;
    virtual void setPostRunCommandHook(PostRunCommandHookFunc func);
    PostRunCommandHookFunc getPostRunCommandHook() const {
        return _postRunCommandHook;
    }
 
 
protected:
    bool isOk(const BSONObj&);
 
    bool isNotMasterErrorString(const BSONElement& e);
 
    BSONObj _countCmd(const std::string& ns, const Query& query, int options, int limit, int skip);
 
    QueryOptions availableOptions();
 
    virtual QueryOptions _lookupAvailableOptions();
 
    virtual void _auth(const BSONObj& params);
 
    bool _authMongoCR(const std::string& dbname,
                      const std::string& username,
                      const std::string& pwd,
                      BSONObj* info,
                      bool digestPassword);
 
    bool _authX509(const std::string& dbname, const std::string& username, BSONObj* info);
 
    RunCommandHookFunc _runCommandHook;
    PostRunCommandHookFunc _postRunCommandHook;
    int _minWireVersion;
    int _maxWireVersion;
 
 
private:
    enum QueryOptions _cachedAvailableOptions;
    bool _haveCachedAvailableOptions;
 
    void _buildGroupObj(const StringData& ns,
                        const StringData& jsreduce,
                        const BSONObj& initial,
                        const Query& query,
                        const StringData& finalize,
                        BSONObjBuilder* groupObj);
 
    void _runGroup(const StringData& ns,
                   const BSONObj& group,
                   const Query& query,
                   std::vector<BSONObj>* output);
 
    std::auto_ptr<DBClientCursor> _legacyCollectionInfo(const std::string& db,
                                                        const BSONObj& filter,
                                                        int batchSize);
};
 
class DBClientWriter;
class WriteOperation;
 
class MONGO_CLIENT_API DBClientBase : public DBClientWithCommands, public DBConnector {
    friend class BulkOperationBuilder;
 
protected:
    static AtomicInt64 ConnectionIdSequence;
    long long _connectionId;  // unique connection id for this connection
    const boost::scoped_ptr<DBClientWriter> _wireProtocolWriter;
    const boost::scoped_ptr<DBClientWriter> _commandWriter;
    WriteConcern _writeConcern;
    int _maxBsonObjectSize;
    int _maxMessageSizeBytes;
    int _maxWriteBatchSize;
    void _write(const std::string& ns,
                const std::vector<WriteOperation*>& writes,
                bool ordered,
                bool bypassDocumentValidation,
                const WriteConcern* writeConcern,
                WriteResult* writeResult);
 
public:
    static const uint64_t INVALID_SOCK_CREATION_TIME;
 
    DBClientBase();
 
    // Required because compiler can't generate a destructor for the _writer
    // as it is an incomplete type.
    virtual ~DBClientBase();
 
    long long getConnectionId() const {
        return _connectionId;
    }
 
    const WriteConcern& getWriteConcern() const {
        return _writeConcern;
    }
    void setWriteConcern(const WriteConcern& w) {
        _writeConcern = w;
    }
 
    void setWireVersions(int minWireVersion, int maxWireVersion) {
        _minWireVersion = minWireVersion;
        _maxWireVersion = maxWireVersion;
    }
 
    virtual int getMinWireVersion() {
        return _minWireVersion;
    }
    virtual int getMaxWireVersion() {
        return _maxWireVersion;
    }
    int getMaxBsonObjectSize() {
        return _maxBsonObjectSize;
    }
    int getMaxMessageSizeBytes() {
        return _maxMessageSizeBytes;
    }
    int getMaxWriteBatchSize() {
        return _maxWriteBatchSize;
    }
 
    virtual std::auto_ptr<DBClientCursor> query(const std::string& ns,
                                                Query query,
                                                int nToReturn = 0,
                                                int nToSkip = 0,
                                                const BSONObj* fieldsToReturn = 0,
                                                int queryOptions = 0,
                                                int batchSize = 0);
 
    virtual void parallelScan(const StringData& ns,
                              int numCursors,
                              std::vector<DBClientCursor*>* cursors,
                              stdx::function<DBClientBase*()> connectionFactory);
 
    virtual std::auto_ptr<DBClientCursor> aggregate(const std::string& ns,
                                                    const BSONObj& pipeline,
                                                    const BSONObj* aggregateOptions = NULL,
                                                    int queryOptions = 0);
 
 
    virtual unsigned long long query(stdx::function<void(const BSONObj&)> f,
                                     const std::string& ns,
                                     Query query,
                                     const BSONObj* fieldsToReturn = 0,
                                     int queryOptions = 0);
 
    virtual unsigned long long query(stdx::function<void(DBClientCursorBatchIterator&)> f,
                                     const std::string& ns,
                                     Query query,
                                     const BSONObj* fieldsToReturn = 0,
                                     int queryOptions = 0);
 
 
    virtual std::auto_ptr<DBClientCursor> getMore(const std::string& ns,
                                                  long long cursorId,
                                                  int nToReturn = 0,
                                                  int options = 0);
 
    virtual void insert(const std::string& ns,
                        BSONObj obj,
                        int flags = 0,
                        const WriteConcern* wc = NULL);
 
    virtual void insert(const std::string& ns,
                        const std::vector<BSONObj>& v,
                        int flags = 0,
                        const WriteConcern* wc = NULL);
 
    virtual void update(const std::string& ns,
                        Query query,
                        BSONObj obj,
                        bool upsert = false,
                        bool multi = false,
                        const WriteConcern* wc = NULL);
 
    virtual void update(
        const std::string& ns, Query query, BSONObj obj, int flags, const WriteConcern* wc = NULL);
 
    BSONObj findAndModify(const StringData& ns,
                          const BSONObj& query,
                          const BSONObj& update,
                          bool upsert = false,
                          bool returnNew = false,
                          const BSONObj& sort = BSONObj(),
                          const BSONObj& fields = BSONObj(),
                          const WriteConcern* wc = NULL,
                          bool bypassDocumentValidation = false);
 
    BSONObj findAndRemove(const StringData& ns,
                          const BSONObj& query,
                          const BSONObj& sort = BSONObj(),
                          const BSONObj& fields = BSONObj(),
                          const WriteConcern* wc = NULL);
 
    virtual BulkOperationBuilder initializeUnorderedBulkOp(const std::string& ns);
 
    virtual BulkOperationBuilder initializeOrderedBulkOp(const std::string& ns);
 
    virtual void remove(const std::string& ns,
                        Query q,
                        bool justOne = 0,
                        const WriteConcern* wc = NULL);
 
    virtual void remove(const std::string& ns,
                        Query query,
                        int flags,
                        const WriteConcern* wc = NULL);
 
    virtual bool isFailed() const = 0;
 
    virtual bool isStillConnected() = 0;
 
    virtual void killCursor(long long cursorID) = 0;
 
    virtual bool callRead(Message& toSend, Message& response) = 0;
    // virtual bool callWrite( Message& toSend , Message& response ) = 0; // TODO: add this if
    // needed
 
    virtual ConnectionString::ConnectionType type() const = 0;
 
    virtual double getSoTimeout() const = 0;
 
    virtual uint64_t getSockCreationMicroSec() const {
        return INVALID_SOCK_CREATION_TIME;
    }
 
    virtual void reset() {}
 
private:
    void _findAndModify(const StringData& ns,
                        const BSONObj& query,
                        const BSONObj& update,
                        const BSONObj& sort,
                        bool returnNew,
                        bool upsert,
                        const BSONObj& fields,
                        const WriteConcern* wc,
                        bool bypassDocumentValidation,
                        BSONObjBuilder* out);
 
};  // DBClientBase
 
class DBClientReplicaSet;
 
class MONGO_CLIENT_API ConnectException : public UserException {
public:
    ConnectException(std::string msg) : UserException(9000, msg) {}
};
 
class MONGO_CLIENT_API DBClientConnection : public DBClientBase {
public:
    using DBClientBase::query;
 
    DBClientConnection(bool _autoReconnect = false,
                       DBClientReplicaSet* cp = 0,
                       double so_timeout = 0);
 
    virtual ~DBClientConnection() {
        _numConnections.fetchAndAdd(-1);
    }
 
    virtual bool connect(const HostAndPort& server, std::string& errmsg);
 
    bool connect(const std::string& server, std::string& errmsg) {
        return connect(HostAndPort(server), errmsg);
    }
 
    void connect(const std::string& serverHostname) {
        std::string errmsg;
        if (!connect(HostAndPort(serverHostname), errmsg))
            throw ConnectException(std::string("can't connect ") + errmsg);
    }
 
    virtual void logout(const std::string& dbname, BSONObj& info);
 
    virtual std::auto_ptr<DBClientCursor> query(const std::string& ns,
                                                Query query = Query(),
                                                int nToReturn = 0,
                                                int nToSkip = 0,
                                                const BSONObj* fieldsToReturn = 0,
                                                int queryOptions = 0,
                                                int batchSize = 0);
 
    virtual unsigned long long query(stdx::function<void(DBClientCursorBatchIterator&)> f,
                                     const std::string& ns,
                                     Query query,
                                     const BSONObj* fieldsToReturn,
                                     int queryOptions);
 
    virtual bool runCommand(const std::string& dbname,
                            const BSONObj& cmd,
                            BSONObj& info,
                            int options = 0);
 
    bool isFailed() const {
        return _failed;
    }
 
    bool isStillConnected() {
        return p ? p->isStillConnected() : true;
    }
 
    MessagingPort& port() {
        verify(p);
        return *p;
    }
 
    std::string toString() const {
        std::stringstream ss;
        ss << _serverString;
        if (!_serverAddrString.empty())
            ss << " (" << _serverAddrString << ")";
        if (_failed)
            ss << " failed";
        return ss.str();
    }
 
    std::string getServerAddress() const {
        return _serverString;
    }
    const HostAndPort& getServerHostAndPort() const {
        return _server;
    }
 
    virtual void killCursor(long long cursorID);
    virtual bool callRead(Message& toSend, Message& response) {
        return call(toSend, response);
    }
    virtual void say(Message& toSend, bool isRetry = false, std::string* actualServer = 0);
    virtual bool recv(Message& m);
    virtual void checkResponse(const char* data,
                               int nReturned,
                               bool* retry = NULL,
                               std::string* host = NULL);
    virtual bool call(Message& toSend,
                      Message& response,
                      bool assertOk = true,
                      std::string* actualServer = 0);
    virtual ConnectionString::ConnectionType type() const {
        return ConnectionString::MASTER;
    }
    void setSoTimeout(double timeout);
    double getSoTimeout() const {
        return _so_timeout;
    }
 
    virtual bool lazySupported() const {
        return true;
    }
 
    static int MONGO_CLIENT_FUNC getNumConnections() {
        return _numConnections.load();
    }
 
    void setParentReplSetName(const std::string& replSetName);
 
    static void MONGO_CLIENT_FUNC setLazyKillCursor(bool lazy) {
        _lazyKillCursor = lazy;
    }
    static bool MONGO_CLIENT_FUNC getLazyKillCursor() {
        return _lazyKillCursor;
    }
 
    uint64_t getSockCreationMicroSec() const;
 
protected:
    virtual void _auth(const BSONObj& params);
    virtual void sayPiggyBack(Message& toSend);
 
    boost::scoped_ptr<MessagingPort> p;
    boost::scoped_ptr<SockAddr> server;
    bool _failed;
    const bool autoReconnect;
    Backoff autoReconnectBackoff;
    HostAndPort _server;            // remember for reconnects
    std::string _serverString;      // server host and port
    std::string _serverAddrString;  // resolved ip of server
    void _checkConnection();
 
    // throws SocketException if in failed state and not reconnecting or if waiting to reconnect
    void checkConnection() {
        if (_failed)
            _checkConnection();
    }
 
    std::map<std::string, BSONObj> authCache;
    double _so_timeout;
    bool _connect(std::string& errmsg);
 
    static AtomicInt32 _numConnections;
    static bool _lazyKillCursor;  // lazy means we piggy back kill cursors on next op
 
#ifdef MONGO_SSL
    SSLManagerInterface* sslManager();
#endif
 
private:
    void handleNotMasterResponse(const BSONElement& elemToCheck);
 
    // Contains the string for the replica set name of the host this is connected to.
    // Should be empty if this connection is not pointing to a replica set member.
    std::string _parentReplSetName;
};
 
MONGO_CLIENT_API bool MONGO_CLIENT_FUNC serverAlive(const std::string& uri);
 
MONGO_CLIENT_API BSONElement MONGO_CLIENT_FUNC getErrField(const BSONObj& result);
MONGO_CLIENT_API bool MONGO_CLIENT_FUNC hasErrField(const BSONObj& result);
 
MONGO_CLIENT_API inline std::ostream& MONGO_CLIENT_FUNC
operator<<(std::ostream& s, const Query& q) {
    return s << q.toString();
}
 
}  // namespace mongo