Mongo.watch.txt

=============
Mongo.watch()
=============

.. default-domain:: mongodb

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

Definition
----------

.. method:: Mongo.watch( pipeline, options )

   *For replica sets and sharded clusters only*

   Opens a :ref:`change stream cursor <changeStreams>` for a replica
   set or a sharded cluster to report on all its non-``system``
   collections across its databases, with the exception of the
   ``admin``, ``local``, and the ``config`` databases.


   .. 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:`Mongo.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:`Mongo.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:`Mongo.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:`Mongo.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:`Mongo.watch()` to look up the most current 
          majority-committed version of the updated document.
          :method:`Mongo.watch()` returns a ``fullDocument`` field with
          the document lookup in addition to the ``updateDescription`` delta.
          
          
   
      * - ``batchSize``
   
        - int
   
        - Optional. Specifies the maximum number of change events to return in each
          batch of the response from the MongoDB cluster. 
          
          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 collation for the change stream cursor.
          
          If omitted, defaults to ``simple`` binary comparison.
          
          
   
      * - ``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` over the change event documents.
      See :doc:`/reference/change-events` for examples of change
      event documents.

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

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

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

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

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

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

Deployment
~~~~~~~~~~

:method:`Mongo.watch()` is available for replica sets and sharded
clusters:

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

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

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

You can only use :method:`Mongo.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:`Mongo.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

.. |watchmethod| replace:: :method:`Mongo.watch()`

.. note:: Resume Token

   .. include:: /includes/extracts/changestream-resume-token-versions-4.2-greater.rst

   Hex Encoded Tokens
   ``````````````````

   .. include:: /includes/extracts/changestream-resume-token-hex-change.rst

   Decode Resume Tokens
   ````````````````````

   .. include:: /includes/note-decode-resume-tokens.rst

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

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

Availability
~~~~~~~~~~~~

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

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

When running with access control, the user must have the
:authaction:`find` and :authaction:`changeStream` privilege actions on
:ref:`all non-systems collections across all
databases<resource-all-but-system-collections>`.. That is, a user must
have a :ref:`role <roles>` that grants the following :ref:`privilege
<privileges>`:

.. code-block:: javascript

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

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

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

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

Example
-------

The following operation in :binary:`~bin.mongosh` opens a
change stream cursor on a replica set. The returned cursor reports on
data changes to all the non-``system`` collections across all databases
except for the ``admin``, ``local``, and the ``config`` databases.

.. code-block:: javascript

   watchCursor = db.getMongo().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