dbclientcursor.h

// file dbclientcursor.h

/*    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 "../db/jsobj.h"
#include "../db/json.h"
#include <stack>

namespace mongo {

    class AScopedConnection;

    class DBClientCursorInterface : boost::noncopyable {
    public:
        virtual ~DBClientCursorInterface() {}
        virtual bool more() = 0;
        virtual BSONObj next() = 0;
        // TODO bring more of the DBClientCursor interface to here
    protected:
        DBClientCursorInterface() {}
    };

    class DBClientCursor : public DBClientCursorInterface {
    public:
        bool more();

        int objsLeftInBatch() const { _assertIfNull(); return _putBack.size() + batch.nReturned - batch.pos; }
        bool moreInCurrentBatch() { return objsLeftInBatch() > 0; }

        BSONObj next();

        void putBack( const BSONObj &o ) { _putBack.push( o.getOwned() ); }

        BSONObj nextSafe() {
            BSONObj o = next();
            if( strcmp(o.firstElementFieldName(), "$err") == 0 ) {
                string s = "nextSafe(): " + o.toString();
                if( logLevel >= 5 )
                    log() << s << endl;
                uasserted(13106, s);
            }
            return o;
        }

        void peek(vector<BSONObj>&, int atMost);

        // Peeks at first element, if exists
        BSONObj peekFirst();

        bool peekError(BSONObj* error = NULL);

        int itcount() {
            int c = 0;
            while ( more() ) {
                next();
                c++;
            }
            return c;
        }

        bool isDead() const { return  !this || cursorId == 0; }

        bool tailable() const { return (opts & QueryOption_CursorTailable) != 0; }

        bool hasResultFlag( int flag ) {
            _assertIfNull();
            return (resultFlags & flag) != 0;
        }

        DBClientCursor( DBClientBase* client, const string &_ns, BSONObj _query, int _nToReturn,
                        int _nToSkip, const BSONObj *_fieldsToReturn, int queryOptions , int bs ) :
            _client(client),
            ns(_ns),
            query(_query),
            nToReturn(_nToReturn),
            haveLimit( _nToReturn > 0 && !(queryOptions & QueryOption_CursorTailable)),
            nToSkip(_nToSkip),
            fieldsToReturn(_fieldsToReturn),
            opts(queryOptions),
            batchSize(bs==1?2:bs),
            cursorId(),
            _ownCursor( true ),
            wasError( false ) {
            _finishConsInit();
        }

        DBClientCursor( DBClientBase* client, const string &_ns, long long _cursorId, int _nToReturn, int options ) :
            _client(client),
            ns(_ns),
            nToReturn( _nToReturn ),
            haveLimit( _nToReturn > 0 && !(options & QueryOption_CursorTailable)),
            opts( options ),
            cursorId(_cursorId),
            _ownCursor( true ) {
            _finishConsInit();
        }

        virtual ~DBClientCursor();

        long long getCursorId() const { return cursorId; }

        void decouple() { _ownCursor = false; }

        void attach( AScopedConnection * conn );

        string originalHost() const { return _originalHost; }

        Message* getMessage(){ return batch.m.get(); }

        bool init();

        void initLazy( bool isRetry = false );
        bool initLazyFinish( bool& retry );

        class Batch : boost::noncopyable { 
            friend class DBClientCursor;
            auto_ptr<Message> m;
            int nReturned;
            int pos;
            const char *data;
        public:
            Batch() : m( new Message() ), nReturned(), pos(), data() { }
        };

    private:
        friend class DBClientBase;
        friend class DBClientConnection;

        int nextBatchSize();
        void _finishConsInit();
        
        Batch batch;
        DBClientBase* _client;
        string _originalHost;
        string ns;
        BSONObj query;
        int nToReturn;
        bool haveLimit;
        int nToSkip;
        const BSONObj *fieldsToReturn;
        int opts;
        int batchSize;
        stack< BSONObj > _putBack;
        int resultFlags;
        long long cursorId;
        bool _ownCursor; // see decouple()
        string _scopedHost;
        string _lazyHost;
        bool wasError;

        void dataReceived() { bool retry; string lazyHost; dataReceived( retry, lazyHost ); }
        void dataReceived( bool& retry, string& lazyHost );
        void requestMore();
        void exhaustReceiveMore(); // for exhaust

        // Don't call from a virtual function
        void _assertIfNull() const { uassert(13348, "connection died", this); }

        // non-copyable , non-assignable
        DBClientCursor( const DBClientCursor& );
        DBClientCursor& operator=( const DBClientCursor& );

        // init pieces
        void _assembleInit( Message& toSend );
    };

    class DBClientCursorBatchIterator {
    public:
        DBClientCursorBatchIterator( DBClientCursor &c ) : _c( c ), _n() {}
        bool moreInCurrentBatch() { return _c.moreInCurrentBatch(); }
        BSONObj nextSafe() {
            massert( 13383, "BatchIterator empty", moreInCurrentBatch() );
            ++_n;
            return _c.nextSafe();
        }
        int n() const { return _n; }
    private:
        DBClientCursor &_c;
        int _n;
    };

} // namespace mongo

#include "undef_macros.h"