change streams

Author: rachel-mack

source/includes/usage-examples/code-snippets/Command.java renamed to source/includes/crud/Command.java

@@ -1,3 +1,5 @@
+package org.example;
+
 import java.util.Arrays;
 import java.util.List;
 
@@ -17,7 +19,7 @@ public class Watch {
     public static void main( String[] args ) {
 
         // Replace the uri string with your MongoDB deployment's connection string
-        String uri = "<connection string URI>";
+        String uri = "<connection string uri>";
 
         try (MongoClient mongoClient = MongoClients.create(uri)) {
             MongoDatabase database = mongoClient.getDatabase("sample_mflix");

source/includes/crud/Watch.java

@@ -1,19 +1,16 @@
-// Performs CRUD operations to generate change events when run with the Watch application
-
-
-package usage.examples;
-
-import java.util.Arrays;
+package org.example;
 
 import org.bson.Document;
-import org.bson.types.ObjectId;
 
 import com.mongodb.MongoException;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
 import com.mongodb.client.result.InsertOneResult;
+import com.mongodb.client.model.Updates;
+import com.mongodb.client.result.UpdateResult;
+import com.mongodb.client.result.DeleteResult;
 
 public class WatchCompanion {
     public static void main(String[] args) {
@@ -27,7 +24,7 @@ public static void main(String[] args) {
             try {
                 // Inserts a sample document into the "movies" collection and print its ID
                 InsertOneResult insertResult = collection.insertOne(new Document("test", "sample movie document"));
-                System.out.println("Success! Inserted document id: " + insertResult.getInsertedId());
+                System.out.println("Inserted document id: " + insertResult.getInsertedId());
 
                 // Updates the sample document and prints the number of modified documents
                 UpdateResult updateResult = collection.updateOne(new Document("test", "sample movie document"), Updates.set("field2", "sample movie document update"));
@@ -36,8 +33,8 @@ public static void main(String[] args) {
                 // Deletes the sample document and prints the number of deleted documents
                 DeleteResult deleteResult = collection.deleteOne(new Document("field2", "sample movie document update"));
                 System.out.println("Deleted " + deleteResult.getDeletedCount() + " document.");
-            
-            // Prints a message if any exceptions occur during the operations
+
+                // Prints a message if any exceptions occur during the operations
             } catch (MongoException me) {
                 System.err.println("Unable to insert, update, or replace due to an error: " + me);
             }

source/includes/usage-examples/code-snippets/WatchCompanion.java renamed to source/includes/crud/WatchCompanion.java

@@ -8,7 +8,7 @@ Open Change Streams
 .. contents:: On this page
    :local:
    :backlinks: none
-   :depth: 1
+   :depth: 2
    :class: singlecol
 
 Overview
@@ -39,6 +39,9 @@ Open a Change Stream
 You can open a change stream to subscribe to specific types of data changes
 and produce change events in your application.
 
+Select a Scope to Watch
+~~~~~~~~~~~~~~~~~~~~~~~
+
 To open a change stream, call the ``watch()`` method on an instance of a
 ``MongoCollection``, ``MongoDatabase``, or ``MongoClient``.
 
@@ -49,16 +52,53 @@ To open a change stream, call the ``watch()`` method on an instance of a
    see the :ref:`<replica-set-oplog>` {+mdb-server+} manual page.
 
 The object on which you call the ``watch()`` method on determines the scope of
-events that the change stream listens for.
+events that the change stream listens for:
+
+- ``MongoCollection.watch()`` monitors a collection.
+- ``MongoDatabase.watch()`` monitors all collections in a database.
+- ``MongoClient.watch()`` monitors all changes in the connected MongoDB deployment.
 
-If you call ``watch()`` on a ``MongoCollection``, the change stream monitors
-a collection.
+Filter the Events
+~~~~~~~~~~~~~~~~~
 
-If you call ``watch()`` on a ``MongoDatabase``, the change stream monitors all
-collections in that database.
+The ``watch()`` method optionally takes an **aggregation pipeline**  which
+consists of an array of **stages** as the first parameter to filter and
+transform the change event output as follows:
+
+.. code-block:: java
 
-If you call ``watch()`` on a ``MongoClient``, the change stream monitors all
-changes in the connected MongoDB deployment.
+   List<Bson> pipeline = Arrays.asList(
+                           Aggregates.match(
+                              Filters.lt("fullDocument.runtime", 15)));
+   ChangeStreamIterable<Document> changeStream = database.watch(pipeline);
+
+Manage the Output
+~~~~~~~~~~~~~~~~~
+
+The ``watch()`` method returns an instance of ``ChangeStreamIterable``, a class
+that offers several methods to access, organize, and traverse the results.
+``ChangeStreamIterable`` also inherits methods from its parent class,
+``MongoIterable`` which implements the core Java interface ``Iterable``.
+
+You can call ``forEach()`` on the ``ChangeStreamIterable`` to handle
+events as they occur, or you can use the ``iterator()`` method which
+returns a ``MongoCursor`` instance that you can use to traverse the results.
+
+You can call methods on the ``MongoCursor`` such as ``hasNext()`` to check
+whether additional results exist, ``next()`` to return the next document
+in the collection, or ``tryNext()``, to immediately return either
+the next available element in the change stream or ``null``. Unlike the
+``MongoCursor`` returned by other queries, a ``MongoCursor`` associated
+with a change stream waits until a change event arrives before
+returning a result from ``next()``. As a result, calls to ``next()``
+using a change stream's ``MongoCursor`` never throw a
+``java.util.NoSuchElementException``.
+
+To configure options for processing the documents returned from the change
+stream, use member methods of the ``ChangeStreamIterable`` object returned
+by ``watch()``. See the link to the ``ChangeStreamIterable`` API
+documentation at the bottom of this example for more details on the
+available methods.
 
 Example
 ~~~~~~~
@@ -90,36 +130,81 @@ An insert operation on the collection produces the following output:
       ...
    }
 
-Watch Example: Full File 
-~~~~~~~~~~~~~~~~~~~~~~~~
+Watch Example: Full Files 
+~~~~~~~~~~~~~~~~~~~~~~~~~
 
 .. include:: /includes/crud/example-intro.rst
 
-This file demonstrates how to open a change stream by using the watch method.
-The watch method takes a pipeline as an argument to filter for only ``"insert"``
-and ``"update"`` events. When an insert or update event occurs on the watched
-collection, a log of the even is printed to the screen.
+This example demonstrates how to open a change stream by using the watch method.
+The ``Watch.java`` file calls the ``watch()`` method with a pipeline as an
+argument to filter for only ``"insert"`` and ``"update"`` events. The
+``WatchCompanion.java`` file inserts, updates and deletes a document.
+
+To use the following examples, run the files in this order:
+
+#. Run the ``Watch.java`` file.
+#. Run the ``WatchCompanion.java`` file.
+
+.. note::
 
-To see output in your terminal:
+   The ``Watch.java`` file will continue running until the ``WatchCompanion.java`` file
 
-#. Run the file in your editor.
-#. Insert or update a document in the ``sample_mflix`` database ``movies`` collection.
+``Watch.java``:
 
-.. tip::
+.. literalinclude:: /includes/crud/Watch.java
+   :language: java
 
-   You can insert or update documents by using :atlas:`Atlas </documents/>` or :mongosh:`mongosh </crud/>`.
+``WatchCompanion.java``:
 
-.. io-code-block::
+.. literalinclude:: /includes/crud/WatchCompanion.java
+   :language: java
 
-   .. input:: /includes/crud/Watch.java
-      :language: java
-      :dedent:
+Full File Example Output
+````````````````````````
 
-   .. output:: 
-      :language: none
-      :visible: false
+The preceding applications will generate the following output:
 
-      Received a change to the collection: ChangeStreamDocument{ operationType=insert, resumeToken={...}, namespace=sample_mflix.movies, ... }
+``Watch.java`` will capture on the  ``insert`` and ``update`` operations are
+printed, since the aggregation pipeline filters out the ``delete`` operation:
+
+.. code-block::
+   :copyable: false
+
+   Received a change to the collection: ChangeStreamDocument{
+     operationType=OperationType{value='insert'},
+     resumeToken={"_data": "825E..."},
+     namespace=sample_mflix.movies,
+     destinationNamespace=null,
+     fullDocument=Document{{_id=5ec3..., test=sample movie document}},
+     documentKey={"_id": {"$oid": "5ec3..."}},
+     clusterTime=Timestamp{...},
+     updateDescription=null,
+     txnNumber=null,
+     lsid=null,
+     wallTime=BsonDateTime{value=1657...}
+   }
+   Received a change to the collection: ChangeStreamDocument{
+     operationType=OperationType{value='update'},
+     resumeToken={"_data": "825E..."},
+     namespace=sample_mflix.movies,
+     destinationNamespace=null,
+     fullDocument=Document{{_id=5ec3..., test=sample movie document, field2=sample movie document update}},
+     documentKey={"_id": {"$oid": "5ec3..."}},
+     clusterTime=Timestamp{...},
+     updateDescription=UpdateDescription{removedFields=[], updatedFields={"field2": "sample movie document update"}},
+     txnNumber=null,
+     lsid=null,
+     wallTime=BsonDateTime{value=1657...}
+   }
+
+``WatchCompanion`` will print all of the operations it completed:
+
+.. code-block::
+   :copyable: false
+
+   Inserted document id: BsonObjectId{value=5ec3...}
+   Updated 1 document.
+   Deleted 1 document.
 
 
 To learn more about the ``watch()`` method, see the following API
@@ -370,3 +455,25 @@ output:
 
 For a list of options, see the `FullDocument <{+api+}/apidocs/mongodb-driver-core/com/mongodb/client/model/changestream/FullDocument.html>`__
 API documentation.
+
+Additional Information
+----------------------
+
+To learn more about the methods and classes used to manage change streams, see the following API documentation:
+
+API Documentation
+~~~~~~~~~~~~~~~~~
+
+- `MongoCollection.watch() <{+api+}/apidocs/mongodb-driver-sync/com/mongodb/client/MongoCollection.html#watch()>`__ 
+- `MongoDatabase.watch() <{+api+}/apidocs/mongodb-driver-sync/com/mongodb/client/MongoDatabase.html#watch()>`__ 
+- `MongoClient.watch() <{+api+}/apidocs/mongodb-driver-sync/com/mongodb/client/MongoClient.html#watch()>`__ 
+- `ChangeStreamIterable <{+api+}/apidocs/mongodb-driver-sync/com/mongodb/client/ChangeStreamIterable.html>`__ 
+- `MongoCursor <{+api+}/apidocs/mongodb-driver-sync/com/mongodb/client/MongoCursor.html>`__ 
+
+Server Manual Entries
+~~~~~~~~~~~~~~~~~~~~~
+
+- :manual:`Change Streams </changeStreams/>`
+- :manual:`Change Events </reference/change-events/>`
+- :manual:`Aggregation Pipeline </reference/operator/aggregation-pipeline/>`
+- :manual:`Aggregation Stages </changeStreams/#modify-change-stream-output>`