dbclient.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 "../pch.h"
#include "../util/net/message.h"
#include "../util/net/message_port.h"
#include "../db/jsobj.h"
#include "../db/json.h"
#include <stack>

namespace mongo {

    enum 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 UpdateOptions {
        UpdateOption_Upsert = 1 << 0,

        UpdateOption_Multi = 1 << 1,

        UpdateOption_Broadcast = 1 << 2
    };

    enum RemoveOptions {
        RemoveOption_JustOne = 1 << 0,

        RemoveOption_Broadcast = 1 << 1
    };

    
    enum InsertOptions {
        InsertOption_ContinueOnError = 1 << 0
    };

    class DBClientBase;

    class ConnectionString {
    public:
        enum ConnectionType { INVALID , MASTER , PAIR , SET , SYNC };

        ConnectionString() {
            _type = INVALID;
        }

        ConnectionString( const HostAndPort& server ) {
            _type = MASTER;
            _servers.push_back( server );
            _finishInit();
        }

        ConnectionString( ConnectionType type , const string& s , const string& setName = "" ) {
            _type = type;
            _setName = setName;
            _fillServers( s );

            switch ( _type ) {
            case MASTER:
                assert( _servers.size() == 1 );
                break;
            case SET:
                assert( _setName.size() );
                assert( _servers.size() >= 1 ); // 1 is ok since we can derive
                break;
            case PAIR:
                assert( _servers.size() == 2 );
                break;
            default:
                assert( _servers.size() > 0 );
            }

            _finishInit();
        }

        ConnectionString( const string& s , ConnectionType favoredMultipleType ) {
            _type = INVALID;
            
            _fillServers( s );
            if ( _type != INVALID ) {
                // set already
            }
            else if ( _servers.size() == 1 ) {
                _type = MASTER;
            }
            else {
                _type = favoredMultipleType;
                assert( _type == SET || _type == SYNC );
            }
            _finishInit();
        }

        bool isValid() const { return _type != INVALID; }

        string toString() const { return _string; }
        
        DBClientBase* connect( string& errmsg, double socketTimeout = 0 ) const;

        string getSetName() const { return _setName; }

        vector<HostAndPort> getServers() const { return _servers; }
        
        ConnectionType type() const { return _type; }

        static ConnectionString parse( const string& url , string& errmsg );

        static string typeToString( ConnectionType type );

    private:

        void _fillServers( string s );
        void _finishInit();

        ConnectionType _type;
        vector<HostAndPort> _servers;
        string _string;
        string _setName;
    };

    enum WriteConcern {
        W_NONE = 0 , // TODO: not every connection type fully supports this
        W_NORMAL = 1
        // TODO SAFE = 2
    };

    class BSONObj;
    class ScopedDbConnection;
    class DBClientCursor;
    class DBClientCursorBatchIterator;

    class Query {
    public:
        BSONObj obj;
        Query() : obj(BSONObj()) { }
        Query(const BSONObj& b) : obj(b) { }
        Query(const string &json) :
            obj(fromjson(json)) { }
        Query(const char * json) :
            obj(fromjson(json)) { }

        Query& sort(const BSONObj& sortPattern);

        Query& sort(const string &field, int asc = 1) { sort( BSON( field << asc ) ); return *this; }

        Query& hint(BSONObj keyPattern);
        Query& hint(const string &jsonKeyPatt) { return hint(fromjson(jsonKeyPatt)); }

        Query& minKey(const BSONObj &val);
        Query& maxKey(const BSONObj &val);

        Query& explain();

        Query& snapshot();

        Query& where(const string &jscode, BSONObj scope);
        Query& where(const string &jscode) { return where(jscode, BSONObj()); }

        bool isComplex( bool * hasDollar = 0 ) const;

        BSONObj getFilter() const;
        BSONObj getSort() const;
        BSONObj getHint() const;
        bool isExplain() const;

        string toString() const;
        operator 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();
        }
    };

#define QUERY(x) mongo::Query( BSON(x) )

    class DBConnector {
    public:
        virtual ~DBConnector() {}
        virtual bool call( Message &toSend, Message &response, bool assertOk=true , string * actualServer = 0 ) = 0;
        virtual void say( Message &toSend, bool isRetry = false ) = 0;
        virtual void sayPiggyBack( Message &toSend ) = 0;
        /* used by QueryOption_Exhaust.  To use that your subclass must implement this. */
        virtual bool recv( Message& m ) { assert(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, string* targetHost = NULL ) {
            if( retry ) *retry = false; if( targetHost ) *targetHost = "";
        }
        virtual bool lazySupported() const = 0;
    };

    class DBClientInterface : boost::noncopyable {
    public:
        virtual auto_ptr<DBClientCursor> query(const 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 string &ns, BSONObj obj , int flags=0) = 0;

        virtual void insert( const string &ns, const vector< BSONObj >& v , int flags=0) = 0;

        virtual void remove( const string &ns , Query query, bool justOne = 0 ) = 0;

        virtual void update( const string &ns , Query query , BSONObj obj , bool upsert = 0 , bool multi = 0 ) = 0;

        virtual ~DBClientInterface() { }

        virtual BSONObj findOne(const string &ns, const Query& query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0);

        void findN(vector<BSONObj>& out, const string&ns, Query query, int nToReturn, int nToSkip = 0, const BSONObj *fieldsToReturn = 0, int queryOptions = 0);

        virtual string getServerAddress() const = 0;

        virtual auto_ptr<DBClientCursor> getMore( const string &ns, long long cursorId, int nToReturn = 0, int options = 0 ) = 0;
    };

    class DBClientWithCommands : public DBClientInterface {
        set<string> _seenIndexes;
    public:
        int _logLevel;

        DBClientWithCommands() : _logLevel(0), _cachedAvailableOptions( (enum QueryOptions)0 ), _haveCachedAvailableOptions(false) { }

        bool simpleCommand(const string &dbname, BSONObj *info, const string &command);

        virtual bool runCommand(const string &dbname, const BSONObj& cmd, BSONObj &info, int options=0);

        virtual bool auth(const string &dbname, const string &username, const string &pwd, string& errmsg, bool digestPassword = true);

        virtual unsigned long long count(const string &ns, const BSONObj& query = BSONObj(), int options=0, int limit=0, int skip=0 );

        string createPasswordDigest( const string &username , const string &clearTextPassword );

        virtual bool isMaster(bool& isMaster, BSONObj *info=0);

        bool createCollection(const string &ns, long long size = 0, bool capped = false, int max = 0, BSONObj *info = 0);

        string getLastError();

        virtual BSONObj getLastErrorDetailed();

        static string getLastErrorString( const BSONObj& res );

        BSONObj getPrevError();

        bool resetError() { return simpleCommand("admin", 0, "reseterror"); }

        virtual bool dropCollection( const string &ns ) {
            string db = nsGetDB( ns );
            string coll = nsGetCollection( ns );
            uassert( 10011 ,  "no collection name", coll.size() );

            BSONObj info;

            bool res = runCommand( db.c_str() , BSON( "drop" << coll ) , info );
            resetIndexCache();
            return res;
        }

        bool repairDatabase(const string &dbname, BSONObj *info = 0) {
            return simpleCommand(dbname, info, "repairDatabase");
        }

        bool copyDatabase(const string &fromdb, const string &todb, const string &fromhost = "", BSONObj *info = 0);

        enum ProfilingLevel {
            ProfileOff = 0,
            ProfileSlow = 1, // log very slow (>100ms) operations
            ProfileAll = 2

        };
        bool setDbProfilingLevel(const string &dbname, ProfilingLevel level, BSONObj *info = 0);
        bool getDbProfilingLevel(const string &dbname, ProfilingLevel& level, BSONObj *info = 0);


        struct MROutput {
            MROutput(const char* collection) : out(BSON("replace" << collection)) {}
            MROutput(const string& collection) : out(BSON("replace" << collection)) {}
            MROutput(const BSONObj& obj) : out(obj) {}

            BSONObj out;
        };
        static MROutput MRInline;

        BSONObj mapreduce(const string &ns, const string &jsmapf, const string &jsreducef, BSONObj query = BSONObj(), MROutput output = MRInline);

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

        bool validate( const 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 string &dbname, const string &jscode);
        template< class T >
        bool eval(const string &dbname, const 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 string &dbname, const 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;
        }

        list<string> getDatabaseNames();

        list<string> getCollectionNames( const string& db );

        bool exists( const string& ns );

        virtual bool ensureIndex( const string &ns , BSONObj keys , bool unique = false, const string &name = "",
                                  bool cache = true, bool background = false, int v = -1 );

        virtual void resetIndexCache();

        virtual auto_ptr<DBClientCursor> getIndexes( const string &ns );

        virtual void dropIndex( const string& ns , BSONObj keys );
        virtual void dropIndex( const string& ns , const string& indexName );

        virtual void dropIndexes( const string& ns );

        virtual void reIndex( const string& ns );

        string genIndexName( const BSONObj& keys );

        virtual bool dropDatabase(const string &dbname, BSONObj *info = 0) {
            bool ret = simpleCommand(dbname, info, "dropDatabase");
            resetIndexCache();
            return ret;
        }

        virtual string toString() = 0;

        string nsGetDB( const string &ns ) {
            string::size_type pos = ns.find( "." );
            if ( pos == string::npos )
                return ns;

            return ns.substr( 0 , pos );
        }

        string nsGetCollection( const string &ns ) {
            string::size_type pos = ns.find( "." );
            if ( pos == string::npos )
                return "";

            return ns.substr( pos + 1 );
        }

    protected:
        bool isOk(const BSONObj&);

        bool isNotMasterErrorString( const BSONElement& e );

        BSONObj _countCmd(const string &ns, const BSONObj& query, int options, int limit, int skip );

        enum QueryOptions availableOptions();

    private:
        enum QueryOptions _cachedAvailableOptions;
        bool _haveCachedAvailableOptions;
    };

    class DBClientBase : public DBClientWithCommands, public DBConnector {
    protected:
        WriteConcern _writeConcern;

    public:
        DBClientBase() {
            _writeConcern = W_NORMAL;
        }

        WriteConcern getWriteConcern() const { return _writeConcern; }
        void setWriteConcern( WriteConcern w ) { _writeConcern = w; }

        virtual auto_ptr<DBClientCursor> query(const string &ns, Query query, int nToReturn = 0, int nToSkip = 0,
                                               const BSONObj *fieldsToReturn = 0, int queryOptions = 0 , int batchSize = 0 );

        virtual auto_ptr<DBClientCursor> getMore( const string &ns, long long cursorId, int nToReturn = 0, int options = 0 );

        virtual void insert( const string &ns , BSONObj obj , int flags=0);

        virtual void insert( const string &ns, const vector< BSONObj >& v , int flags=0);

        virtual void remove( const string &ns , Query q , bool justOne = 0 );

        virtual void update( const string &ns , Query query , BSONObj obj , bool upsert = false , bool multi = false );

        virtual bool isFailed() const = 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;

    }; // DBClientBase

    class DBClientReplicaSet;

    class ConnectException : public UserException {
    public:
        ConnectException(string msg) : UserException(9000,msg) { }
    };

    class DBClientConnection : public DBClientBase {
    public:
        DBClientConnection(bool _autoReconnect=false, DBClientReplicaSet* cp=0, double so_timeout=0) :
            clientSet(cp), _failed(false), autoReconnect(_autoReconnect), lastReconnectTry(0), _so_timeout(so_timeout) {
            _numConnections++;
        }

        virtual ~DBClientConnection() {
            _numConnections--;
        }

        virtual bool connect(const char * hostname, string& errmsg) {
            // TODO: remove this method
            HostAndPort t( hostname );
            return connect( t , errmsg );
        }

        virtual bool connect(const HostAndPort& server, string& errmsg);

        void connect(const string& serverHostname) {
            string errmsg;
            if( !connect(HostAndPort(serverHostname), errmsg) )
                throw ConnectException(string("can't connect ") + errmsg);
        }

        virtual bool auth(const string &dbname, const string &username, const string &pwd, string& errmsg, bool digestPassword = true);

        virtual auto_ptr<DBClientCursor> query(const string &ns, Query query=Query(), int nToReturn = 0, int nToSkip = 0,
                                               const BSONObj *fieldsToReturn = 0, int queryOptions = 0 , int batchSize = 0 ) {
            checkConnection();
            return DBClientBase::query( ns, query, nToReturn, nToSkip, fieldsToReturn, queryOptions , batchSize );
        }

        unsigned long long query( boost::function<void(const BSONObj&)> f, const string& ns, Query query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0);
        unsigned long long query( boost::function<void(DBClientCursorBatchIterator&)> f, const string& ns, Query query, const BSONObj *fieldsToReturn = 0, int queryOptions = 0);

        virtual bool runCommand(const string &dbname, const BSONObj& cmd, BSONObj &info, int options=0);

        bool isFailed() const { return _failed; }

        MessagingPort& port() { assert(p); return *p; }

        string toStringLong() const {
            stringstream ss;
            ss << _serverString;
            if ( _failed ) ss << " failed";
            return ss.str();
        }

        string toString() { return _serverString; }

        string getServerAddress() const { return _serverString; }

        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 );
        virtual bool recv( Message& m );
        virtual void checkResponse( const char *data, int nReturned, bool* retry = NULL, string* host = NULL );
        virtual bool call( Message &toSend, Message &response, bool assertOk = true , string * actualServer = 0 );
        virtual ConnectionString::ConnectionType type() const { return ConnectionString::MASTER; }
        void setSoTimeout(double to) { _so_timeout = to; }
        double getSoTimeout() const { return _so_timeout; }

        virtual bool lazySupported() const { return true; }

        static int getNumConnections() {
            return _numConnections;
        }
        
        static void setLazyKillCursor( bool lazy ) { _lazyKillCursor = lazy; }
        static bool getLazyKillCursor() { return _lazyKillCursor; }
        
    protected:
        friend class SyncClusterConnection;
        virtual void sayPiggyBack( Message &toSend );

        DBClientReplicaSet *clientSet;
        boost::scoped_ptr<MessagingPort> p;
        boost::scoped_ptr<SockAddr> server;
        bool _failed;
        const bool autoReconnect;
        time_t lastReconnectTry;
        HostAndPort _server; // remember for reconnects
        string _serverString;
        void _checkConnection();

        // throws SocketException if in failed state and not reconnecting or if waiting to reconnect
        void checkConnection() { if( _failed ) _checkConnection(); }

        map< string, pair<string,string> > authCache;
        double _so_timeout;
        bool _connect( string& errmsg );

        static AtomicUInt _numConnections;
        static bool _lazyKillCursor; // lazy means we piggy back kill cursors on next op

#ifdef MONGO_SSL
        static SSLManager* sslManager();
        static SSLManager* _sslManager;
#endif
    };

    bool serverAlive( const string &uri );

    DBClientBase * createDirectClient();

    BSONElement getErrField( const BSONObj& result );
    bool hasErrField( const BSONObj& result );

} // namespace mongo

#include "dbclientcursor.h"
#include "dbclient_rs.h"
#include "undef_macros.h"