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()
        );

        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()
        );

        BSONObj findAndRemove(
            const StringData& ns,
            const BSONObj& query,
            const BSONObj& sort = BSONObj(),
            const BSONObj& fields = BSONObj()
        );

        bool eval(const std::string &dbname, const std::string &jscode, BSONObj& info, BSONElement& retValue, BSONObj *args = 0);

        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 */
        bool eval(const std::string &dbname, const std::string &jscode);
        template< class T >
        bool eval(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 >
        bool eval(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
        );

        void _findAndModify(
            const StringData& ns,
            const BSONObj& query,
            const BSONObj& update,
            const BSONObj& sort,
            bool returnNew,
            bool upsert,
            const BSONObj& fields,
            BSONObjBuilder* out
        );

        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,
            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;
        }

        int getMinWireVersion() { return _minWireVersion; }
        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
        );

        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() {}

    }; // 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) :
            clientSet(cp), _failed(false), autoReconnect(_autoReconnect), autoReconnectBackoff(1000, 2000), _so_timeout(so_timeout) {
            _numConnections.fetchAndAdd(1);
        }

        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 setReplSetClientCallback(DBClientReplicaSet* rsClient);

        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 );

        DBClientReplicaSet *clientSet;
        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
    };

    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