add tailable option to find() to use a tailable cursor

Author: Mike Dirolf

pymongo/collection.py

@@ -305,7 +305,7 @@ def _fields_list_to_dict(self, fields):
  return as_dict
 
  def find(self, spec=None, fields=None, skip=0, limit=0,
- slave_okay=None, timeout=True, snapshot=False,
+ slave_okay=None, timeout=True, snapshot=False, tailable=False,
  _sock=None, _must_use_master=False):
  """Query the database.
 
@@ -342,9 +342,15 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
  - `snapshot` (optional): if True, snapshot mode will be used for this
  query. Snapshot mode assures no duplicates are returned, or objects
  missed, which were present at both the start and end of the query's
- execution. For details, see the wiki_
-
- .. _wiki: http://www.mongodb.org/display/DOCS/How+to+do+Snapshotting+in+the+Mongo+Database
+ execution. For details, see the `wiki
+ <http://www.mongodb.org/display/DOCS/How+to+do+Snapshotting+in+the+Mongo+Database>`_.
+ - `tailable` (optional): the result of this find call will be a
+ tailable cursor - tailable cursors aren't closed when the last data
+ is retrieved but are kept open and the cursors location marks the
+ final document's position. if more data is received iteration of
+ the cursor will continue from the last document received. For
+ details, see the `wiki
+ <http://www.mongodb.org/display/DOCS/Tailable+Cursors>`_.
  """
  if spec is None:
  spec = SON()
@@ -370,14 +376,17 @@ def find(self, spec=None, fields=None, skip=0, limit=0,
  raise TypeError("timeout must be an instance of bool")
  if not isinstance(snapshot, types.BooleanType):
  raise TypeError("snapshot must be an instance of bool")
+ if not isinstance(tailable, types.BooleanType):
+ raise TypeError("tailable must be an instance of bool")
 
  if fields is not None:
  if not fields:
  fields = ["_id"]
  fields = self._fields_list_to_dict(fields)
 
- return Cursor(self, spec, fields, skip, limit, slave_okay, timeout, snapshot,
- _sock=_sock, _must_use_master=_must_use_master)
+ return Cursor(self, spec, fields, skip, limit, slave_okay, timeout,
+ tailable, snapshot, _sock=_sock,
+ _must_use_master=_must_use_master)
 
  def count(self):
  """Get the number of documents in this collection.

pymongo/cursor.py

@@ -37,7 +37,8 @@ class Cursor(object):
  """
 
  def __init__(self, collection, spec, fields, skip, limit, slave_okay,
- timeout, snapshot=False, _sock=None, _must_use_master=False):
+ timeout, tailable, snapshot=False,
+ _sock=None, _must_use_master=False):
  """Create a new cursor.
 
  Should not be called directly by application developers.
@@ -49,6 +50,7 @@ def __init__(self, collection, spec, fields, skip, limit, slave_okay,
  self.__limit = limit
  self.__slave_okay = slave_okay
  self.__timeout = timeout
+ self.__tailable = tailable
  self.__snapshot = snapshot
  self.__ordering = None
  self.__explain = False
@@ -99,7 +101,7 @@ def clone(self):
  """
  copy = Cursor(self.__collection, self.__spec, self.__fields,
  self.__skip, self.__limit, self.__slave_okay,
- self.__timeout, self.__snapshot)
+ self.__timeout, self.__tailable, self.__snapshot)
  copy.__ordering = self.__ordering
  copy.__explain = self.__explain
  copy.__hint = self.__hint
@@ -141,6 +143,8 @@ def __query_options(self):
  """Get the 4 byte query options string to use for this query.
  """
  options = 0
+ if self.__tailable:
+ options |= _QUERY_OPTIONS["tailable_cursor"]
  if self.__slave_okay:
  options |= _QUERY_OPTIONS["slave_okay"]
  if not self.__timeout:
@@ -381,7 +385,11 @@ def __send_message(self, operation, message):
 
  response = pymongo.Connection._unpack_response(response, self.__id)
  self.__id = response["cursor_id"]
- assert response["starting_from"] == self.__retrieved
+
+ # starting from doesn't get set on getmore's for tailable cursors
+ if not self.__tailable:
+ assert response["starting_from"] == self.__retrieved
+
  self.__retrieved += response["number_returned"]
  self.__data = response["data"]
 

test/test_cursor.py

@@ -555,6 +555,36 @@ def set_coll():
 
  self.assertRaises(AttributeError, set_coll)
 
+ def test_tailable(self):
+ db = self.db
+ db.drop_collection("test")
+
+
+ cursor = db.test.find(tailable=True)
+
+ db.test.insert({"x": 1})
+ count = 0
+ for doc in cursor:
+ count += 1
+ self.assertEqual(1, doc["x"])
+ self.assertEqual(1, count)
+
+ db.test.insert({"x": 2})
+ count = 0
+ for doc in cursor:
+ count += 1
+ self.assertEqual(2, doc["x"])
+ self.assertEqual(1, count)
+
+ db.test.insert({"x": 3})
+ count = 0
+ for doc in cursor:
+ count += 1
+ self.assertEqual(3, doc["x"])
+ self.assertEqual(1, count)
+
+ self.assertEqual(3, db.test.count())
+
 
 if __name__ == "__main__":
  unittest.main()