SERVER-4150 enable setting client cursor's slaveReadTill field

3 files changed, 47 insertions, 8 deletions

diff --git a/src/mongo/db/clientcursor.h b/src/mongo/db/clientcursor.h
index 0022a9e6190..501cbbb8261 100644
--- a/src/mongo/db/clientcursor.h
+++ b/src/mongo/db/clientcursor.h
@@ -362,6 +362,9 @@ namespace mongo {
         void setDoingDeletes( bool doingDeletes ) {_doingDeletes = doingDeletes; }
 
         void slaveReadTill( const OpTime& t ) { _slaveReadTill = t; }
+        
+        /** Just for testing. */
+        OpTime getSlaveReadTill() const { return _slaveReadTill; }
 
     public: // static methods
 
diff --git a/src/mongo/db/ops/query.cpp b/src/mongo/db/ops/query.cpp
index 605037ba5e6..ffefd02d40b 100644
--- a/src/mongo/db/ops/query.cpp
+++ b/src/mongo/db/ops/query.cpp
@@ -605,8 +605,9 @@ namespace mongo {
         bool wouldSaveClientCursor() const { return _wouldSaveClientCursor; }
 
         void finishForOplogReplay( ClientCursor * cc ) {
-            if ( _oplogReplay && ! _slaveReadTill.isNull() )
+            if ( _oplogReplay && ! _slaveReadTill.isNull() ) {
                 cc->slaveReadTill( _slaveReadTill );
+            }
 
         }
 
@@ -1221,13 +1222,13 @@ namespace mongo {
 //                    log() << "idx: " << cursor->indexKeyPattern() << " obj: " << cursor->current() << endl;
                     
 
-                    // This should happen after matching?
-//                    if ( pq.hasOption( QueryOption_OplogReplay ) ) {
-//                        BSONElement e = js["ts"];
-//                        if ( e.type() == Date || e.type() == Timestamp ) {
-//                            slaveReadTill = e._opTime();
-//                        }
-//                    }
+                    if ( pq.hasOption( QueryOption_OplogReplay ) ) {
+                        BSONObj current = cursor->current();
+                        BSONElement e = current["ts"];
+                        if ( e.type() == Date || e.type() == Timestamp ) {
+                            slaveReadTill = e._opTime();
+                        }
+                    }
                     
                     if ( !cursor->supportGetMore() || pq.isExplain() ) {
                         if ( queryResponseBuilder.enoughTotalResults() ) {
@@ -1266,6 +1267,8 @@ namespace mongo {
                         ccPointer->updateLocation();
                     }
                     ccPointer->originalMessage = m;
+                    if ( pq.hasOption( QueryOption_OplogReplay ) && !slaveReadTill.isNull() )
+                        ccPointer->slaveReadTill( slaveReadTill );
                     ccPointer.release();
                     // undo unlimited timeout
                 }
diff --git a/src/mongo/dbtests/querytests.cpp b/src/mongo/dbtests/querytests.cpp
index 3aff1c2c56e..760664b3462 100644
--- a/src/mongo/dbtests/querytests.cpp
+++ b/src/mongo/dbtests/querytests.cpp
@@ -414,6 +414,38 @@ namespace QueryTests {
         }
     };
 
+    class OplogReplaySlaveReadTill : public ClientBase {
+    public:
+        ~OplogReplaySlaveReadTill() {
+            client().dropCollection( "unittests.querytests.OplogReplaySlaveReadTill" );
+        }
+        void run() {
+            const char *ns = "unittests.querytests.OplogReplaySlaveReadTill";
+            BSONObj info;
+            client().runCommand( "unittests",
+                                BSON( "create" << "querytests.OplogReplaySlaveReadTill" <<
+                                     "capped" << true << "size" << 8192 ),
+                                info );
+            Date_t one = OpTime::now().asDate();
+            Date_t two = OpTime::now().asDate();
+            Date_t three = OpTime::now().asDate();
+            insert( ns, BSON( "ts" << one ) );
+            insert( ns, BSON( "ts" << two ) );
+            insert( ns, BSON( "ts" << three ) );
+            auto_ptr<DBClientCursor> c =
+            client().query( ns, QUERY( "ts" << GTE << two ).hint( BSON( "$natural" << 1 ) ),
+                           0, 0, 0, QueryOption_OplogReplay | QueryOption_CursorTailable );
+            ASSERT( c->more() );
+            ASSERT_EQUALS( two, c->next()["ts"].Date() );
+            long long cursorId = c->getCursorId();
+            
+            dblock lk;
+            Client::Context ctx( "unittests.querytests.OplogReplaySlaveReadTill" );
+            ClientCursor::Pointer clientCursor( cursorId );
+            ASSERT_EQUALS( three.millis, clientCursor.c()->getSlaveReadTill().asDate() );
+        }
+    };
+
     class BasicCount : public ClientBase {
     public:
         ~BasicCount() {
@@ -1435,6 +1467,7 @@ namespace QueryTests {
             add< TailCappedOnly >();
             add< TailableQueryOnId >();
             add< OplogReplayMode >();
+            add< OplogReplaySlaveReadTill >();
             add< ArrayId >();
             add< UnderscoreNs >();
             add< EmptyFieldSpec >();