PYTHON-1662 Add ChangeStream.try_next API

Author: ShaneHarvey

doc/api/pymongo/change_stream.rst

@@ -1,5 +1,5 @@
-:mod:`change_stream` -- Watch changes on a collection
-=====================================================
+:mod:`change_stream` -- Watch changes on a collection, database, or cluster
+===========================================================================
 
 .. automodule:: pymongo.change_stream
    :members:

doc/changelog.rst

@@ -84,6 +84,9 @@ Changes in Version 3.8.0.dev0
   string.
 - Add the ``filter`` parameter to
   :meth:`~pymongo.database.Database.list_collection_names`.
+- Changes can now be requested from a ``ChangeStream`` cursor without blocking
+  indefinitely using the new
+  :meth:`pymongo.change_stream.ChangeStream.try_next` method.
 
 Issues Resolved
 ...............

pymongo/change_stream.py

@@ -12,7 +12,7 @@
 # implied.  See the License for the specific language governing
 # permissions and limitations under the License.
 
-"""ChangeStream cursor to iterate over changes on a collection."""
+"""Watch changes on a collection, a database, or the entire cluster."""
 
 import copy
 
@@ -41,14 +41,12 @@ class ChangeStream(object):
     """The internal abstract base class for change stream cursors.
 
     Should not be called directly by application developers. Use 
-    :meth:pymongo.collection.Collection.watch,
-    :meth:pymongo.database.Database.watch, or
-    :meth:pymongo.mongo_client.MongoClient.watch instead.
-
-    Defines the interface for change streams. Should be subclassed to
-    implement the `ChangeStream._create_cursor` abstract method, and
-    the `ChangeStream._database`and ChangeStream._aggregation_target`
-    abstract properties.
+    :meth:`pymongo.collection.Collection.watch`,
+    :meth:`pymongo.database.Database.watch`, or
+    :meth:`pymongo.mongo_client.MongoClient.watch` instead.
+
+    .. versionadded:: 3.6
+    .. mongodoc:: changeStreams
     """
     def __init__(self, target, pipeline, full_document, resume_after,
                  max_await_time_ms, batch_size, collation,
@@ -176,34 +174,97 @@ def next(self):
         """Advance the cursor.
 
         This method blocks until the next change document is returned or an
-        unrecoverable error is raised.
+        unrecoverable error is raised. This method is used when iterating over
+        all changes in the cursor. For example::
+
+            try:
+                with db.collection.watch(
+                        [{'$match': {'operationType': 'insert'}}]) as stream:
+                    for insert_change in stream:
+                        print(insert_change)
+            except pymongo.errors.PyMongoError:
+                # The ChangeStream encountered an unrecoverable error or the
+                # resume attempt failed to recreate the cursor.
+                logging.error('...')
 
         Raises :exc:`StopIteration` if this ChangeStream is closed.
         """
-        while True:
-            try:
-                change = self._cursor.next()
-            except ConnectionFailure:
-                self._resume()
-                continue
-            except OperationFailure as exc:
-                if exc.code in _NON_RESUMABLE_GETMORE_ERRORS:
-                    raise
-                self._resume()
-                continue
-            try:
-                resume_token = change['_id']
-            except KeyError:
-                self.close()
-                raise InvalidOperation(
-                    "Cannot provide resume functionality when the resume "
-                    "token is missing.")
-            self._resume_token = copy.copy(resume_token)
-            self._start_at_operation_time = None
-            return change
+        while self.alive:
+            doc = self.try_next()
+            if doc is not None:
+                return doc
+
+        raise StopIteration
 
     __next__ = next
 
+    @property
+    def alive(self):
+        """Does this cursor have the potential to return more data?
+
+        .. note:: Even if :attr:`alive` is ``True``, :meth:`next` can raise
+            :exc:`StopIteration` and :meth:`try_next` can return ``None``.
+
+        .. versionadded:: 3.8
+        """
+        return self._cursor.alive
+
+    def try_next(self):
+        """Advance the cursor without blocking indefinitely.
+
+        This method returns the next change document without waiting
+        indefinitely for the next change. For example::
+
+            with db.collection.watch() as stream:
+                while stream.alive:
+                    change = stream.try_next()
+                    if change is not None:
+                        print(change)
+                    elif stream.alive:
+                        # We end up here when there are no recent changes.
+                        # Sleep for a while to avoid flooding the server with
+                        # getMore requests when no changes are available.
+                        time.sleep(10)
+
+        If no change document is cached locally then this method runs a single
+        getMore command. If the getMore yields any documents, the next
+        document is returned, otherwise, if the getMore returns no documents
+        (because there have been no changes) then ``None`` is returned.
+
+        :Returns:
+          The next change document or ``None`` when no document is available
+          after running a single getMore or when the cursor is closed.
+
+        .. versionadded:: 3.8
+        """
+        # Attempt to get the next change with at most one getMore and at most
+        # one resume attempt.
+        try:
+            change = self._cursor._try_next(True)
+        except ConnectionFailure:
+            self._resume()
+            change = self._cursor._try_next(False)
+        except OperationFailure as exc:
+            if exc.code in _NON_RESUMABLE_GETMORE_ERRORS:
+                raise
+            self._resume()
+            change = self._cursor._try_next(False)
+
+        # No changes are available.
+        if change is None:
+            return None
+
+        try:
+            resume_token = change['_id']
+        except KeyError:
+            self.close()
+            raise InvalidOperation(
+                "Cannot provide resume functionality when the resume "
+                "token is missing.")
+        self._resume_token = copy.copy(resume_token)
+        self._start_at_operation_time = None
+        return change
+
     def __enter__(self):
         return self
 
@@ -212,13 +273,12 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 
 class CollectionChangeStream(ChangeStream):
-    """Class for creating a change stream on a collection.
+    """A change stream that watches changes on a single collection.
 
     Should not be called directly by application developers. Use
     helper method :meth:`pymongo.collection.Collection.watch` instead.
 
-    .. versionadded: 3.6
-    .. mongodoc:: changeStreams
+    .. versionadded:: 3.7
     """
     @property
     def _aggregation_target(self):
@@ -230,13 +290,12 @@ def _database(self):
 
 
 class DatabaseChangeStream(ChangeStream):
-    """Class for creating a change stream on all collections in a database.
+    """A change stream that watches changes on all collections in a database.
 
     Should not be called directly by application developers. Use
     helper method :meth:`pymongo.database.Database.watch` instead.
 
-    .. versionadded: 3.7
-    .. mongodoc:: changeStreams
+    .. versionadded:: 3.7
     """
     @property
     def _aggregation_target(self):
@@ -248,13 +307,12 @@ def _database(self):
 
 
 class ClusterChangeStream(DatabaseChangeStream):
-    """Class for creating a change stream on all collections on a cluster.
+    """A change stream that watches changes on all collections in the cluster.
 
     Should not be called directly by application developers. Use
     helper method :meth:`pymongo.mongo_client.MongoClient.watch` instead.
 
-    .. versionadded: 3.7
-    .. mongodoc:: changeStreams
+    .. versionadded:: 3.7
     """
     def _pipeline_options(self):
         options = super(ClusterChangeStream, self)._pipeline_options()

pymongo/command_cursor.py

@@ -285,15 +285,24 @@ def __iter__(self):
     def next(self):
         """Advance the cursor."""
         # Block until a document is returnable.
-        while not len(self.__data) and not self.__killed:
+        while self.alive:
+            doc = self._try_next(True)
+            if doc is not None:
+                return doc
+
+        raise StopIteration
+
+    __next__ = next
+
+    def _try_next(self, get_more_allowed):
+        """Advance the cursor blocking for at most one getMore command."""
+        if not len(self.__data) and not self.__killed and get_more_allowed:
             self._refresh()
         if len(self.__data):
             coll = self.__collection
             return coll.database._fix_outgoing(self.__data.popleft(), coll)
         else:
-            raise StopIteration
-
-    __next__ = next
+            return None
 
     def __enter__(self):
         return self