db.collection.watch.txt

=====================
db.collection.watch()
=====================

.. default-domain:: mongodb

.. contents:: On this page
   :local:
   :backlinks: none
   :depth: 2
   :class: singlecol

Definition
----------

.. method:: db.collection.watch( pipeline, options )

   .. |dbcommand| replace:: :dbcommand:`aggregate` command with the
      :pipeline:`$changeStream` aggregation stage 
   .. include:: /includes/fact-mongosh-shell-method-alt

   *For replica sets and sharded clusters only*

   Opens a :ref:`change stream cursor <changeStreams>` on the 
   collection. 

   .. list-table::
      :header-rows: 1
      :widths: 20 20 80
   
      * - Parameter
        - Type
        - Description

      * - ``pipeline``
        - array
        - Optional. An :ref:`aggregation-pipeline` consisting of one
          or more of the following aggregation stages:

          .. include:: /includes/extracts/changestream-available-pipeline-stages.rst

          Specify a pipeline to filter/modify the change events output.

          .. include:: /includes/extracts/4.2-changes-change-stream-modification-error.rst

      * - ``options``
        - document
        - Optional. Additional options that modify the behavior of
          :method:`~db.collection.watch()`.

   The ``options`` document can contain the following fields and values:

   .. list-table::
      :header-rows: 1
      :widths: 20 20 80
   
      * - Field
        - Type
        - Description
   
      * - ``resumeAfter``
        - document
        - Optional. Directs :method:`~db.collection.watch` to attempt resuming
          notifications starting after the operation specified in the resume
          token. 
          
          Each change stream event document includes a resume token as the
          ``_id`` field. Pass the *entire* ``_id`` field of the change event
          document that represents the operation you want to resume after.
          
          ``resumeAfter`` is mutually exclusive with ``startAfter`` and
          ``startAtOperationTime``.
          
          
   
      * - ``startAfter``
   
        - document
        - Optional. Directs :method:`~db.collection.watch` to attempt starting a new
          change stream after the operation specified in the resume token.
          Allows notifications to resume after an invalidate event.
                  
          Each change stream event document includes a resume token as the
          ``_id`` field. Pass the *entire* ``_id`` field of the change event
          document that represents the operation you want to resume after.
                  
          ``startAfter`` is mutually exclusive with ``resumeAfter`` and
          ``startAtOperationTime``.

   
      * - ``fullDocument``
   
        - string
        - Optional. By default, :method:`~db.collection.watch()` returns the delta of
          those fields modified by an update operation, instead of the entire
          updated document.
          
          Set ``fullDocument`` to ``"updateLookup"`` to direct
          :method:`~db.collection.watch()` to look up the most current 
          majority-committed version of the updated document.
          :method:`~db.collection.watch()` returns a ``fullDocument`` field with
          the document lookup in addition to the ``updateDescription`` delta.
          
          .. include:: /includes/change-stream-pre-and-post-images-full-document.rst


      * - ``fullDocumentBeforeChange``   

        - string
        - Optional.

          .. include:: /includes/change-stream-pre-and-post-images-full-document-before-change.rst


      * - ``batchSize``
   
        - int
        - Optional. The maximum number of documents within each batch returned in a change stream. 
          By default, :method:`~db.collection.watch()` has an initial batch size of ``101`` documents and a maximum size 
          of 16 mebibytes (MiB) for each subsequent batch. This option can enforce a smaller 
          limit than 16 MiB, but not a larger one. If you set ``batchSize`` to a limit that
          results in batches larger than 16 MiB, this option has no effect.
          
          Has the same functionality as :method:`cursor.batchSize()`.
          
          
      * - ``maxAwaitTimeMS``
   
        - int
        - Optional. The maximum amount of time in milliseconds the server waits for new
          data changes to report to the change stream cursor before returning an
          empty batch.
          
          Defaults to ``1000`` milliseconds.
 
   
      * - ``collation``
   
        - document
        - Optional. Pass a :ref:`collation document <collation-document-fields>`
          to specify a collation for the change stream cursor.
          
          Defaults to ``simple`` binary comparison if omitted. 
          
      * - ``showExpandedEvents``

        - boolean 
        - Optional. Starting in MongoDB 6.0, change streams support change 
          notifications for DDL events, like the :ref:`createIndexes <change-event-createIndexes>` 
          and :ref:`dropIndexes <change-event-dropIndexes>` events. To include 
          expanded events in a change stream, create the change stream cursor 
          using the ``showExpandedEvents`` option.

          .. versionadded:: 6.0


      * - ``startAtOperationTime``
   
        - Timestamp
        - Optional. The starting point for the change stream. If the specified starting
          point is in the past, it must be in the time range of the oplog. To
          check the time range of the oplog, see
          :method:`rs.printReplicationInfo()`.
          
          ``startAtOperationTime`` is mutually exclusive with ``resumeAfter``
          and ``startAfter``.

   :returns:
      A :term:`cursor` that remains open as long as a connection to the
      MongoDB deployment remains open *and* the collection exists.
      See :doc:`/reference/change-events` for examples of change
      event documents.

   .. seealso::
      
      :method:`db.watch()` and :method:`Mongo.watch()`

Compatibility 
-------------

This method is available in deployments hosted in the following environments:

.. include:: /includes/fact-environments-atlas-only.rst

.. include:: /includes/fact-environments-atlas-support-all.rst

.. include:: /includes/fact-environments-onprem-only.rst

Availability
------------

Deployment
~~~~~~~~~~

:method:`db.collection.watch()` is available for replica set and
sharded cluster deployments :

- For a replica set, you can issue :method:`db.collection.watch()` on
  any data-bearing member.

- For a sharded cluster, you must issue :method:`db.collection.watch()`
  on a :binary:`~bin.mongos` instance.

Storage Engine
~~~~~~~~~~~~~~

You can only use :method:`db.collection.watch()` with the :ref:`Wired
Tiger storage engine <storage-wiredtiger>`.

Read Concern ``majority`` Support
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. include:: /includes/extracts/changestream-rc-majority-4.2.rst

Behavior
--------

- :method:`db.collection.watch()` only notifies on data changes that have
  persisted to a majority of data-bearing members.

- .. include:: /includes/extracts/changestream-cursor-open.rst

Resumability
~~~~~~~~~~~~

.. include:: /includes/extracts/changestream-resume.rst

.. include:: /includes/change-stream/resume-after

.. |watchmethod| replace:: :method:`db.collection.watch()`


Full Document Lookup of Update Operations
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. include:: /includes/extracts/changestream-full-document-lookup.rst

Access Control
--------------

When running with access control, the user must have the
:authaction:`find` and :authaction:`changeStream` privilege actions on
the :ref:`collection resource <resource-document>`. That is, a user must
have a :ref:`role <roles>` that grants the following :ref:`privilege
<privileges>`:

.. code-block:: javascript

   { resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }


The built-in :authrole:`read` role provides the appropriate
privileges.

Cursor Iteration
----------------

.. include:: /includes/fact-multiple-cursor-monitors.rst

Examples
--------

Open a Change Stream
~~~~~~~~~~~~~~~~~~~~

The following operation opens a change stream cursor against the 
``data.sensors`` collection:

.. code-block:: javascript

   watchCursor = db.getSiblingDB("data").sensors.watch()

Iterate the cursor to check for new events. Use the
:method:`cursor.isClosed()` method with the :method:`cursor.tryNext()`
method to ensure the loop only exits if the change stream cursor is
closed *and* there are no objects remaining in the latest batch:

.. code-block:: javascript

   while (!watchCursor.isClosed()) {
     let next = watchCursor.tryNext()
     while (next !== null) {
       printjson(next);
       next = watchCursor.tryNext()
     }
   }

For complete documentation on change stream output, see
:ref:`change-stream-output`.

.. include:: /includes/isExhausted-no-change-streams.rst

Change Stream with Full Document Update Lookup
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

Set the ``fullDocument`` option to ``"updateLookup"`` to direct the
change stream cursor to lookup the most current majority-committed
version of the document associated to an update change stream event.

The following operation opens a change stream cursor against
the ``data.sensors`` collection using the 
``fullDocument : "updateLookup"`` option.

.. code-block:: javascript

   watchCursor = db.getSiblingDB("data").sensors.watch(
      [],
      { fullDocument : "updateLookup" }
   )

Iterate the cursor to check for new events. Use the
:method:`cursor.isClosed()` method with the :method:`cursor.tryNext()`
method to ensure the loop only exits if the change stream cursor is
closed *and* there are no objects remaining in the latest batch:

.. code-block:: javascript

   while (!watchCursor.isClosed()) {
     let next = watchCursor.tryNext()
     while (next !== null) {
       printjson(next);
       next = watchCursor.tryNext()
     }
   }

For any update operation, the change event returns the result of the
document lookup in the ``fullDocument`` field.

For an example of the full document update output, see  :ref:`change
stream update event <change-streams-update-event>`.

For complete documentation on change stream output, see
:ref:`change-stream-output`.

.. _db.collection.watch-change-streams-pre-and-post-images-example:

Change Streams with Document Pre- and Post-Images
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. include:: /includes/change-stream-pre-and-post-images-introduction.rst

.. include:: /includes/change-stream-pre-and-post-images-additional-information.rst

Create Collection
`````````````````

Create a ``temperatureSensor`` collection that has
:ref:`changeStreamPreAndPostImages
<db.createCollection.changeStreamPreAndPostImages>` enabled:

.. code-block:: javascript

   db.createCollection(
      "temperatureSensor",
      { changeStreamPreAndPostImages: { enabled: true } }
   )

Populate the ``temperatureSensor`` collection with temperature readings:

.. code-block:: javascript

   db.temperatureSensor.insertMany( [
      { "_id" : 0, "reading" : 26.1 },
      { "_id" : 1, "reading" : 25.9 },
      { "_id" : 2, "reading" : 24.3 },
      { "_id" : 3, "reading" : 22.4 },
      { "_id" : 4, "reading" : 24.6 }
   ] )

The following sections show change stream examples for document pre- and
post-images that use the ``temperatureSensor`` collection.

Change Stream with Document Pre-Image
`````````````````````````````````````

You use the ``fullDocumentBeforeChange: "whenAvailable"`` setting to
output the document pre-image, if available. The pre-image is the
document before it was replaced, updated, or deleted. There is no
pre-image for an inserted document.

The following example creates a change stream cursor for the
``temperatureSensor`` collection using ``fullDocumentBeforeChange:
"whenAvailable"``:

.. code-block:: javascript

   watchCursorFullDocumentBeforeChange = db.temperatureSensor.watch(
      [],
      { fullDocumentBeforeChange: "whenAvailable" }
   )

The following example uses the cursor to check for new change stream
events:

.. code-block:: javascript

   while ( !watchCursorFullDocumentBeforeChange.isClosed() ) {
      if ( watchCursorFullDocumentBeforeChange.hasNext() ) {
         printjson( watchCursorFullDocumentBeforeChange.next() );
      }
   }

.. include:: /includes/change-stream-pre-and-post-images-example-cursor-methods.rst

The following example updates the ``reading`` field for a
``temperatureSensor`` document:

.. code-block:: javascript

   db.temperatureSensor.updateOne(
      { _id: 2 },
      { $set: { reading: 22.1 } }
   )

After the ``temperatureSensor`` document is updated, the change event
outputs the document pre-image in the ``fullDocumentBeforeChange``
field. The pre-image contains the ``temperatureSensor`` document
``reading`` field before it was updated. For example:

.. code-block:: javascript
   :copyable: false

   {
      "_id" : {
         "_data" : "82624B21...",
         "_typeBits" : BinData(0,"QA==")
      },
      "operationType" : "update",
      "clusterTime" : Timestamp(1649090957, 1),
      "ns" : {
         "db" : "test",
         "coll" : "temperatureSensor"
      },
      "documentKey" : {
         "_id" : 2
      },
      "updateDescription" : {
         "updatedFields" : {
            "reading" : 22.1
         },
         "removedFields" : [ ],
         "truncatedArrays" : [ ]
      },
      "fullDocumentBeforeChange" : {
         "_id" : 2,
         "reading" : 24.3
      }
   }

.. include:: /includes/change-stream-pre-and-post-images-output.rst

Change Stream with Document Post-Image
``````````````````````````````````````

You use the ``fullDocument: "whenAvailable"`` setting to output the
document post-image, if available. The post-image is the document after
it was inserted, replaced, or updated. There is no post-image for a
deleted document.

The following example creates a change stream cursor for the
``temperatureSensor`` collection using ``fullDocument:
"whenAvailable"``:

.. code-block:: javascript

   watchCursorFullDocument = db.temperatureSensor.watch(
      [],
      { fullDocument: "whenAvailable" }
   )

The following example uses the cursor to check for new change stream
events:

.. code-block:: javascript

   while ( !watchCursorFullDocument.isClosed() ) {
      if ( watchCursorFullDocument.hasNext() ) {
         printjson( watchCursorFullDocument.next() );
      }
   }

.. include:: /includes/change-stream-pre-and-post-images-example-cursor-methods.rst

The following example updates the ``reading`` field for a
``temperatureSensor`` document:

.. code-block:: javascript

   db.temperatureSensor.updateOne(
      { _id: 1 },
      { $set: { reading: 29.5 } }
   )

After the ``temperatureSensor`` document is updated, the change event
outputs the document post-image in the ``fullDocument`` field. The
post-image contains the ``temperatureSensor`` document ``reading`` field
after it was updated. For example:

.. code-block:: javascript
   :copyable: false

   {
      "_id" : {
         "_data" : "8262474D...",
         "_typeBits" : BinData(0,"QA==")
      },
      "operationType" : "update",
      "clusterTime" : Timestamp(1648840090, 1),
      "fullDocument" : {
         "_id" : 1,
         "reading" : 29.5
      },
      "ns" : {
         "db" : "test",
         "coll" : "temperatureSensor"
      },
      "documentKey" : {
         "_id" : 1
      },
      "updateDescription" : {
         "updatedFields" : {
            "reading" : 29.5
         },
         "removedFields" : [ ],
         "truncatedArrays" : [ ]
      }
   }

.. include:: /includes/change-stream-pre-and-post-images-output.rst

Change Stream with Aggregation Pipeline Filter
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. note::

   .. include:: /includes/extracts/4.2-changes-change-stream-modification-error.rst

The following operation opens a change stream cursor against the 
``data.sensors`` collection using an aggregation pipeline to
filter only ``insert`` events:

.. code-block:: javascript

   watchCursor = db.getSiblingDB("data").sensors.watch(
      [
         { $match : {"operationType" : "insert" } }
      ]
   )

Iterate the cursor to check for new events. Use the
:method:`cursor.isClosed()` method with the :method:`cursor.hasNext()`
method to ensure the loop only exits if the change stream cursor is
closed *and* there are no objects remaining in the latest batch:

.. code-block:: javascript

   while (!watchCursor.isClosed()){
      if (watchCursor.hasNext()){
         printjson(watchCursor.next());
      }
   }

The change stream cursor only returns change events where the 
``operationType`` is ``insert``. For complete documentation on
change stream output, see :ref:`change-stream-output`. 

Resuming a Change Stream
~~~~~~~~~~~~~~~~~~~~~~~~

Every  document returned by a change stream cursor includes a resume
token as the ``_id`` field. To resume a change stream, pass the entire
``_id`` document of the change event you want to resume from to
either the ``resumeAfter`` or ``startAfter`` option of
:method:`~db.collection.watch()`.

The following operation resumes a change stream cursor against the 
``data.sensors`` collection using a resume token. This
assumes that the operation that generated the resume token has not
rolled off the cluster's oplog. 

.. code-block:: javascript

   let watchCursor = db.getSiblingDB("data").sensors.watch();
   let firstChange;

   while (!watchCursor.isClosed()) {
      if (watchCursor.hasNext()) {
        firstChange = watchCursor.next();
        break;
      }
   }

   watchCursor.close();

   let resumeToken = firstChange._id;

   resumedWatchCursor = db.getSiblingDB("data").sensors.watch(
   [],
      { resumeAfter : resumeToken }
   )

Iterate the cursor to check for new events. Use the
:method:`cursor.isClosed()` method with the :method:`cursor.hasNext()`
method to ensure the loop only exits if the change stream cursor is
closed *and* there are no objects remaining in the latest batch:


.. code-block:: javascript

   while (!resumedWatchCursor.isClosed()){
      if (resumedWatchCursor.hasNext()){
         print(resumedWatchCursor.next());
      }
   }

See :ref:`change-stream-resume` for complete documentation on
resuming a change stream.