JAVA-448: Tailable Cursor and awaitdata

Author: agirbal

src/main/com/mongodb/DBApiLayer.java

@@ -407,15 +407,31 @@ public DBObject next(){
         }
 
         public boolean hasNext(){
-            while ( true ){
-                if ( _cur.hasNext() )
-                    return true;
-
+            boolean hasNext = _cur.hasNext();
+            while ( !hasNext ) {
                 if ( ! _curResult.hasGetMore( _options ) )
                     return false;
 
                 _advance();
+                hasNext = _cur.hasNext();
+                
+                if (!hasNext) {
+                    if ( ( _options & Bytes.QUERYOPTION_AWAITDATA ) == 0 ) {
+                        // dont block waiting for data if no await
+                        return false;
+                    } else {
+                        // if await, driver should block until data is available
+                        // if server does not support await, driver must sleep to avoid busy loop
+                        if ((_curResult._flags & Bytes.RESULTFLAG_AWAITCAPABLE) == 0) {
+                            try {
+                                Thread.sleep(500);
+                            } catch (Exception e) {
+                            }
+                        }
+                    }
+                }
             }
+            return hasNext;
         }
 
         private void _advance(){

src/main/com/mongodb/DBCursor.java

@@ -450,17 +450,7 @@ private DBObject _next()
 
         _check();
 
-        _cur = null;
-        if ((_options & Bytes.QUERYOPTION_TAILABLE) > 0) {
-            try {
-                _cur = _it.next();
-            } catch (NoSuchElementException e){
-                // Hacky way of handling this but should be OK for now...
-                return null;
-            }
-        } else {
-            _cur = _it.next();
-        }
+        _cur = _it.next();
         _num++;
 
         if ( _keysWanted != null && _keysWanted.keySet().size() > 0 ){
@@ -744,7 +734,8 @@ public String toString() {
         if (addr != null)
             sb.append(", addr=").append(addr);
 
-        sb.append(", readPreference=").append( _readPref.toString() );
+        if (_readPref != null)
+            sb.append(", readPreference=").append( _readPref.toString() );
         return sb.toString();
     }
 

src/main/com/mongodb/Response.java

@@ -95,11 +95,7 @@ public boolean hasGetMore( int queryOptions ){
         if ( ( queryOptions & Bytes.QUERYOPTION_TAILABLE ) == 0 )
             return false;
 
-        // have a tailable cursor
-
-        if ( ( _flags & Bytes.RESULTFLAG_AWAITCAPABLE ) > 0 && ( queryOptions & Bytes.QUERYOPTION_AWAITDATA ) > 0 )
-            return true;
-
+        // have a tailable cursor, it is always possible to call get more
         return true;
     }
 

src/test/com/mongodb/DBCursorTest.java

@@ -19,6 +19,8 @@
 import java.io.IOException;
 import java.util.Iterator;
 
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import org.testng.annotations.Test;
 
 import com.mongodb.util.TestCase;
@@ -80,20 +82,80 @@ public void testOptions() {
         assertEquals(0, dbCursor.getOptions());
     }
 
-    @Test
+//    @Test
+//    public void testTailable() {
+//        DBCollection c = _db.createCollection( "tailableTest", new BasicDBObject( "capped", true ).append( "size", 10000));
+//        DBCursor cursor = c.find( ).addOption( Bytes.QUERYOPTION_TAILABLE );
+//
+//        long start = System.currentTimeMillis();
+//        System.err.println( "[ " + start + " ] Has Next?" +  cursor.hasNext());
+//        cursor.next();
+//        long end = System.currentTimeMillis();
+//        System.err.println(  "[ " + end + " ] Tailable next returned." );
+//        assertLess(start - end, 100);
+//        c.drop();
+//    }
+
+    @Test//(enabled = false)
     public void testTailable() {
-        DBCollection c = _db.createCollection( "tailableTest", new BasicDBObject( "capped", true ).append( "size", 10000));
-        DBCursor cursor = c.find( ).addOption( Bytes.QUERYOPTION_TAILABLE );
-
-        long start = System.currentTimeMillis();
-        System.err.println( "[ " + start + " ] Has Next?" +  cursor.hasNext());
-        cursor.next();
-        long end = System.currentTimeMillis();
-        System.err.println(  "[ " + end + " ] Tailable next returned." );
-        assertLess(start - end, 100);
+        DBCollection c = _db.getCollection("tail1");
         c.drop();
+        _db.createCollection("tail1", new BasicDBObject("capped", true).append("size", 10000));
+        for (int i = 0; i < 10; i++) {
+            c.save(new BasicDBObject("x", i));
+        }
+
+        DBCursor cur = c.find().sort(new BasicDBObject("$natural", 1)).addOption(Bytes.QUERYOPTION_TAILABLE);
+
+        while (cur.hasNext()) {
+            cur.next();
+            //do nothing...
+        }
+
+        assert (!cur.hasNext());
+        c.save(new BasicDBObject("x", 12));
+        assert (cur.hasNext());
+        assertNotNull(cur.next());
+        assert (!cur.hasNext());
     }
 
+    @Test//(enabled = false)
+    public void testTailableAwait() {
+        DBCollection c = _db.getCollection("tail1");
+        c.drop();
+        _db.createCollection("tail1", new BasicDBObject("capped", true).append("size", 10000));
+        for (int i = 0; i < 10; i++) {
+            c.save(new BasicDBObject("x", i));
+        }
+
+        final DBCursor cur = c.find().sort(new BasicDBObject("$natural", 1)).addOption(Bytes.QUERYOPTION_TAILABLE | Bytes.QUERYOPTION_AWAITDATA);
+        Thread t = new Thread(new Runnable() {
+            public void run() {
+                // the following call will block on the last hasNext
+                int i = 0;
+                while (cur.hasNext()) {
+                    cur.next();
+                    if (++i > 10)
+                        break;
+                }
+            }
+        });
+        t.start();
+
+        try {
+            Thread.sleep(5000);
+        } catch (InterruptedException ex) {
+        }
+        assert (t.isAlive());
+
+        // this doc should unblock thread
+        c.save(new BasicDBObject("x", 12));
+        try {
+            t.join(100);
+        } catch (InterruptedException ex) {
+        }
+    }
+    
     @Test
     public void testBig2(){
         DBCollection c = _db.getCollection("big2");