changeStreams.txt

.. facet::
   :name: programming_language
   :values: c, csharp, go, java, javascript/typescript, kotlin, php, python, ruby, swift

.. _changeStreams:
.. _collection_watch:

==============
Change Streams
==============

.. default-domain:: mongodb

.. facet::
   :name: genre
   :values: reference

.. contents:: On this page
   :local:
   :backlinks: none
   :depth: 1
   :class: twocols

.. meta::
   :description: Change streams code examples for how to access real-time data changes in MongoDB
   :keywords: database triggers, code example, node.js, motor, coroutine, java sync, swift sync, swift async, compass

Change streams allow applications to access real-time data changes
without the prior complexity and risk of manually tailing the :term:`oplog`.
Applications can use change streams to subscribe to all data changes on
a single collection, a database, or an entire deployment, and
immediately react to them. Because change streams use the aggregation
framework, applications can also filter for specific changes or
transform the notifications at will.

.. |change-streams| replace:: change streams

.. include:: /includes/change-streams-optimization.rst

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

Change streams are available for :ref:`replica sets <replication>` and
:ref:`sharded clusters <sharding-background>`:

- **Storage Engine.** 

  The replica sets and sharded clusters must use the :ref:`WiredTiger
  <storage-wiredtiger>` storage engine. Change streams can also be used
  on deployments that employ MongoDB's
  :ref:`encryption-at-rest<encrypted-storage-engine>` feature.
  
- **Replica Set Protocol Version.** 

  The replica sets and sharded clusters must use replica set protocol
  version 1 (:rsconf:`pv1 <protocolVersion>`).

- **Read Concern "majority" Enablement.**  

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

Stable API Support
~~~~~~~~~~~~~~~~~~

.. include:: /includes/stable-api/change-stream-support.rst

Connect
--------

Connections for a change stream can either use DNS seed lists
with the ``+srv`` connection option or by listing the servers individually
in the connection string.

If the driver loses the connection to a change stream or the connection goes
down, it attempts to reestablish a connection to the change stream through
another node in the cluster that has a matching 
:ref:`read preference <connections-read-preference>`. If the driver cannot find
a node with the correct read preference, it throws an exception.

For more information, see :ref:`Connection String URI Format <mongodb-uri>`.

.. _changeStreams-watch-deployment:

Watch a Collection, Database, or Deployment
-------------------------------------------

You can open change streams against:

.. list-table::
   :header-rows: 1
   :widths: 15 85

   * - Target
     - Description

   * - A collection

     - You can open a change stream cursor for a single collection
       (except ``system`` collections, or any collections in the
       ``admin``, ``local``, and ``config`` databases).

       The examples on this page use the MongoDB drivers to open and
       work with a change stream cursor for a single collection. See
       also the :binary:`~bin.mongosh` method
       :method:`db.collection.watch()`.

   * - A database

     - You can open a change stream cursor for a single database (excluding 
       ``admin``, ``local``, and ``config`` database) to watch for changes to 
       all its non-system collections.

       For the MongoDB driver method, refer to your driver
       documentation. See also the :binary:`~bin.mongosh` method
       :method:`db.watch()`.


   * - A deployment

     - You can open a change stream cursor for a deployment (either a replica 
       set or a sharded cluster) to watch for changes to all non-system 
       collections across all databases except for ``admin``, ``local``, and 
       ``config``.

       For the MongoDB driver method, refer to your driver
       documentation. See also the :binary:`~bin.mongosh` method
       :method:`Mongo.watch()`.

.. note:: Change Stream Examples

   The examples on this page use the MongoDB drivers to illustrate how
   to open a change stream cursor for a collection and work with the
   change stream cursor.

Change Stream Performance Considerations
----------------------------------------

If the amount of active change streams opened against a database exceeds the 
:ref:`connection pool size <connection-pool-overview>`, you may 
experience notification latency. Each change stream uses a connection 
and a :ref:`getMore <manual-reference-commands-getMore>` 
operation on the change stream for the period of time that it waits for the next event. 
To avoid any latency issues, you should ensure that the pool size is greater than the 
number of opened change streams. For details see the :ref:`maxPoolSize<maxpoolsize-cp-setting>` setting.

Sharded Cluster Considerations
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

When a change stream is opened on a sharded cluster:

- The :binary:`mongos` creates individual change streams on **each
  shard**. This behavior occurs regardless of whether the change stream
  targets a particular shard key range.

- When the ``mongos`` receives change stream results, it sorts and
  filters those results. If needed, the ``mongos`` also performs a
  ``fullDocument`` lookup.

For best performance, limit the use of :pipeline:`$lookup` queries in
change streams.

.. _open-change-stream:

Open A Change Stream
--------------------

To open a change stream:

- For a replica set, you can issue the open change stream operation
  from any of the data-bearing members.

- For a sharded cluster, you must issue the open change stream
  operation from the :binary:`~bin.mongos`.

The following example opens a change stream for a collection and
iterates over the cursor to retrieve the change stream documents.
[#start-time]_

----------

.. |arrow| unicode:: U+27A4

|arrow| Use the **Select your language** drop-down menu in the
upper-right to set the language of the examples on this page.

----------

.. tabs-selector:: drivers

.. tabs-drivers::

   .. tab::
      :tabid: python
         
      The Python examples below assume that you have :driver:`connected to a MongoDB replica set and have accessed a database
      </pymongo>` that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/test_examples.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: motor
      
      The examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <https://motor.readthedocs.io/en/stable/tutorial-asyncio.html#creating-a-client>`__
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/test_examples_motor.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: java-sync
           
      The Java examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <http://mongodb.github.io/mongo-java-driver/3.6/driver/tutorials/databases-collections/>`__
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/DocumentationSamples.java
         :language: java
         :dedent: 8
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: kotlin-coroutine
           
      The Kotlin examples below assume that you are connected to a MongoDB replica set and can access a database
      that contains the ``inventory`` collection. To learn more about completing these tasks, see the
      :driver:`Kotlin Coroutine Driver Databases and Collections </kotlin/coroutine/current/fundamentals/databases-collections/>` guide.

      .. literalinclude:: /driver-examples/kotlin_examples.kt
         :language: kotlin
         :dedent:
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab:: 
      :tabid: csharp
      
      The C# examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <http://mongodb.github.io/mongo-csharp-driver/2.4/getting_started/quick_tour/#make-a-connection/>`__
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/ChangeStreamExamples.cs
         :language: csharp
         :dedent: 12
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: c
      
      The C examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <https://www.mongodb.com/docs/languages/c/c-driver/current/databases-collections/#access-a-database>`__
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/test-mongoc-sample-commands.c
         :language: c
         :dedent: 3
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: nodejs

      The Node.js examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <https://mongodb.github.io/node-mongodb-native/api-generated/mongoclient.html#connect>`__
      that contains an ``inventory`` collection.

      The following example uses stream to process the change events.

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

      Alternatively, you can also use iterator to process the change events:

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 1 Alternative
         :end-before: End Changestream Example 1 Alternative

      ChangeStream extends `EventEmitter 
      <https://mongodb.github.io/node-mongodb-native/5.7/classes/TypedEventEmitter.html>`__.

   .. tab::
      :tabid: php

      The examples below assume that you have `connected to a MongoDB replica  set and have accessed a database
      <https://www.mongodb.com/docs/php-library/current/reference/method/MongoDBClient__construct/>`__
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/DocumentationExamplesTest.php
         :language: php
         :dedent: 8
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: ruby

      The examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <https://www.mongodb.com/docs/ruby-driver/current/reference/create-client/>`__
      that contains an ``inventory`` collection.


      .. literalinclude:: /driver-examples/change_stream_examples_spec.rb
         :language: ruby
         :dedent: 8
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: go

      The Go examples below assume that you have `connected to a MongoDB replica set and have accessed a database
      <https://godoc.org/go.mongodb.org/mongo-driver/mongo#NewClient/>`__
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/go_examples.go
         :language: go
         :dedent: 2
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: swift-sync
   
      The Swift (Sync) examples below assume that you have
      `connected to a MongoDB replica set and have accessed a
      database 
      <https://mongodb.github.io/mongo-swift-driver/docs/current/MongoSwiftSync/Classes/MongoClient.html>`__ 
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/swiftSync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

   .. tab::
      :tabid: swift-async
    
      The Swift (Async) examples below assume that you have
      `connected to a MongoDB replica set and have accessed a
      database 
      <https://mongodb.github.io/mongo-swift-driver/docs/current/MongoSwift/Classes/MongoClient.html>`__ 
      that contains an ``inventory`` collection.

      .. literalinclude:: /driver-examples/swiftAsync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 1
         :end-before: End Changestream Example 1

To retrieve the :ref:`data change event <change-stream-output>` from
the cursor, iterate the change stream cursor. For information on the
change stream event, see :ref:`change-stream-output`.

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

.. note::

   The lifecycle of an unclosed cursor is language-dependent.

.. [#start-time]

   You can specify a ``startAtOperationTime`` to open the cursor at a particular 
   point in time. If the specified starting point is in the past, it must be in 
   the time range of the oplog.

.. _change-stream-modify-output:

Modify Change Stream Output
---------------------------

----------

|arrow| Use the **Select your language** drop-down menu in the
upper-right to set the language of the examples on this page.

----------

.. tabs-drivers::

   .. tab:: 
      :tabid: python

      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/test_examples.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4


   .. tab::
      :tabid: motor

      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/test_examples_motor.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

   .. tab::
      :tabid: java-sync

      .. include:: /includes/fact-change-streams-modify-output.rst

      .. code-block:: java

         MongoClient mongoClient = MongoClients.create("mongodb://<username>:<password>@<host>:<port>");

         // Select the MongoDB database and collection to open the change stream against

         MongoDatabase db = mongoClient.getDatabase("myTargetDatabase");

         MongoCollection<Document> collection = db.getCollection("myTargetCollection");

         // Create $match pipeline stage.
         List<Bson> pipeline = singletonList(Aggregates.match(Filters.or(
            Document.parse("{'fullDocument.username': 'alice'}"),
            Filters.in("operationType", asList("delete")))));

         // Create the change stream cursor, passing the pipeline to the
         // collection.watch() method

         MongoCursor<Document> cursor = collection.watch(pipeline).iterator();

      The ``pipeline`` list includes a single :pipeline:`$match` stage that
      filters for any operations that meet one or both of the following criteria:
      
      - ``username`` value is ``alice``
      - ``operationType`` value is ``delete``

      Passing the ``pipeline`` to the :method:`~db.collection.watch()` method directs the
      change stream to return notifications after passing them through the
      specified ``pipeline``.

   .. tab::
      :tabid: kotlin-coroutine
           
      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/kotlin_examples.kt
         :language: kotlin
         :dedent:
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

      The ``pipeline`` list includes a single :pipeline:`$match` stage that
      filters for any operations that meet one or both of the following criteria:
      
      - ``username`` value is ``alice``
      - ``operationType`` value is ``delete``

      Passing the ``pipeline`` to the :method:`~db.collection.watch()` method directs the
      change stream to return notifications after passing them through the
      specified ``pipeline``.

   .. tab::
      :tabid: csharp      
      
      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/ChangeStreamExamples.cs
         :language: csharp
         :dedent: 16
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4
      
   .. tab::
      :tabid: c      
      
      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/test-mongoc-sample-commands.c
         :language: c
         :dedent: 3
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

   .. tab::
      :tabid: nodejs

      .. include:: /includes/fact-change-streams-modify-output.rst

      The following example uses stream to process the change events.

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

      Alternatively, you can also use iterator to process the change events:

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 4 Alternative
         :end-before: End Changestream Example 4 Alternative


   .. tab::
      :tabid: php      
      
      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/DocumentationExamplesTest.php
         :language: php
         :dedent: 8
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

   .. tab::
      :tabid: ruby      
      
      .. include:: /includes/fact-change-streams-modify-output.rst

   .. tab::
      :tabid: go
         
      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/go_examples.go
         :language: go
         :dedent: 2
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

   .. tab::
      :tabid: swift-sync

      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/swiftSync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4

   .. tab::
      :tabid: swift-async
         
      .. include:: /includes/fact-change-streams-modify-output.rst

      .. literalinclude:: /driver-examples/swiftAsync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 4
         :end-before: End Changestream Example 4


.. tip::

   The :ref:`_id <change-stream-event-id>` field of the change stream
   event document act as the :ref:`resume token
   <change-stream-resume>`. Do not use the pipeline to modify or remove
   the change stream event's ``_id`` field. 

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

   See :ref:`change-stream-output` for more information on the change stream
   response document format.

.. _change-streams-updateLookup:

Lookup Full Document for Update Operations
------------------------------------------

By default, change streams only return the delta of fields during
the update operation. However, you can configure the change stream
to return the most current majority-committed version of the updated
document.

----------

|arrow| Use the **Select your language** drop-down menu in the
upper-right to set the language of the examples on this page.

----------

.. tabs-drivers::

   .. tab::
      :tabid: python

      To return the most current majority-committed version of the updated
      document, pass ``full_document='updateLookup'`` to the
      :method:`db.collection.watch()` method.

      In the example below, all update operations notifications
      include a ``full_document`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/test_examples.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: motor

      To return the most current majority-committed version of the updated
      document, pass ``full_document='updateLookup'`` to the
      :method:`db.collection.watch()` method.

      In the example below, all update operations notifications
      include a ```full_document`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/test_examples_motor.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: java-sync

      To return the most current majority-committed version of the updated
      document, pass ``FullDocument.UPDATE_LOOKUP``  to the
      ``db.collection.watch.fullDocument()`` method.

      In the example below, all update operations notifications
      include a ``FullDocument`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/DocumentationSamples.java
         :language: java
         :dedent: 8
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: kotlin-coroutine

      To return the most current majority-committed version of the updated
      document, pass ``FullDocument.UPDATE_LOOKUP``  to the
      `ChangeStreamFlow.fullDocument() <{+java-api-docs+}/mongodb-driver-kotlin-coroutine/mongodb-driver-kotlin-coroutine/com.mongodb.kotlin.client.coroutine/-change-stream-flow/full-document.html>`__ method.

      In the example below, all update operations notifications
      include a ``FullDocument`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/kotlin_examples.kt
         :language: kotlin
         :dedent:
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: c

      To return the most current majority-committed version of the updated
      document, pass the ``"fullDocument"`` option with the ``"updateLookup"`` value to the
      ``mongoc_collection_watch`` method.

      In the example below, all update operations notifications
      include a ``fullDocument`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/test-mongoc-sample-commands.c
         :language: c
         :dedent: 3
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: csharp

      To return the most current majority-committed version of the updated
      document, pass ``"FullDocument = ChangeStreamFullDocumentOption.UpdateLookup"`` to the
      :method:`db.collection.watch()` method.

      In the example below, all update operations notifications
      include a ``FullDocument`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/ChangeStreamExamples.cs
         :language: csharp
         :dedent: 12
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2


   .. tab::
      :tabid: nodejs

      To return the most current majority-committed version of the updated
      document, pass ``{ fullDocument: 'updateLookup' }`` to the
      :method:`db.collection.watch()` method.

      In the example below, all update operations notifications
      include a ``fullDocument`` field that represents the *current*
      version of the document affected by the update operation.

      The following example uses stream to process the change events.

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

      Alternatively, you can also use iterator to process the change events:

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 2 Alternative
         :end-before: End Changestream Example 2 Alternative

   .. tab::
      :tabid: php

      To return the most current
      majority-committed version of the updated document, pass
      ``"fullDocument' => \MongoDB\Operation\ChangeStreamCommand::FULL_DOCUMENT_UPDATE_LOOKUP"``
      to the :method:`db.watch()` method.

      In the example below, all update operations notifications
      include a ``fullDocument`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/DocumentationExamplesTest.php
         :language: php
         :dedent: 8
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: ruby
      
      To return the most current majority-committed version of the updated
      document, pass ``full_document: 'updateLookup'`` to the
      :method:`db.watch()` method.

      In the example below, all update operations notifications
      include a ``full_document`` field that represents the *current*
      version of the document affected by the update operation.

      .. literalinclude:: /driver-examples/change_stream_examples_spec.rb
         :language: ruby
         :dedent: 8
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: go

      To return the most current majority-committed version of the
      updated document, ``SetFullDocument(options.UpdateLookup)``
      change stream option.

      .. literalinclude:: /driver-examples/go_examples.go
         :language: go
         :dedent: 2
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: swift-sync

      To return the most current majority-committed version of
      the updated document, pass ``options:
      ChangeStreamOptions(fullDocument: .updateLookup)`` to the
      ``watch()`` method.

      .. literalinclude:: /driver-examples/swiftSync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

   .. tab::
      :tabid: swift-async

      To return the most current majority-committed version of
      the updated document, pass ``options:
      ChangeStreamOptions(fullDocument: .updateLookup)`` to the
      ``watch()`` method.

      .. literalinclude:: /driver-examples/swiftAsync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 2
         :end-before: End Changestream Example 2

.. note::

   If there are one or more majority-committed operations that modified
   the updated document *after* the update operation but *before* the
   lookup, the full document returned may differ significantly from the
   document at the time of the update operation.

   However, the deltas included in the change stream document always
   correctly describe the watched collection changes that applied to
   that change stream event.

   The ``fullDocument`` field for an update event may be missing if one 
   of the following is true:

   - If the document is deleted or if the collection is dropped in 
     between the update and the lookup.
   - If the update changes the values for at least one of the fields in
     that collection's shard key.

   See :ref:`change-stream-output` for more information on the change
   stream response document format.

.. _change-stream-resume:

Resume a Change Stream
----------------------

Change streams are resumable by specifying a resume token to either
:ref:`resumeAfter <change-stream-resume-after>` or
:ref:`startAfter <change-stream-start-after>` when opening the cursor.

.. _change-stream-resume-after:

``resumeAfter`` for Change Streams
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

You can resume a change stream after a specific event by passing a resume token
to ``resumeAfter`` when opening the cursor. 

See :ref:`change-stream-resume-token` for more information on the resume token.

.. important::

   - The oplog must have enough history to locate the operation
     associated with the token or the timestamp, if the timestamp is in
     the past.

   - .. include:: /includes/extracts/changestream-invalid-events.rst

.. tabs-drivers::

   .. tab::
      :tabid: python

      You can use the ``resume_after`` modifier to resume
      notifications after the operation specified in the resume
      token. The ``resume_after`` modifier takes a value that must
      resolve to a resume token, e.g. ``resume_token`` in the
      example below.


      .. literalinclude:: /driver-examples/test_examples.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: motor

      You can use the ``resume_after`` modifier to resume
      notifications after the operation specified in the resume
      token. The ``resume_after`` modifier takes a value that must
      resolve to a resume token, e.g. ``resume_token`` in the
      example below.

      .. literalinclude:: /driver-examples/test_examples_motor.py
         :language: python
         :dedent: 12
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: java-sync
   
      You can use the ``resumeAfter()`` method to resume
      notifications after the operation specified in the resume
      token. The ``resumeAfter()`` method takes a value that must
      resolve to a resume token, e.g. ``resumeToken`` in the
      example below.

      .. literalinclude:: /driver-examples/DocumentationSamples.java
         :language: java
         :dedent: 8
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: kotlin-coroutine
   
      You can use the `ChangeStreamFlow.resumeAfter()
      <{+java-api-docs+}/mongodb-driver-kotlin-coroutine/mongodb-driver-kotlin-coroutine/com.mongodb.kotlin.client.coroutine/-change-stream-flow/resume-after.html>`__
      method to resume notifications after the operation specified in the resume
      token. The ``resumeAfter()`` method takes a value that must
      resolve to a resume token, such as the ``resumeToken`` variable in the
      example below.

      .. literalinclude:: /driver-examples/kotlin_examples.kt
         :language: kotlin
         :dedent:
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: csharp
   
      In the example below, the ``resumeToken`` is retrieved from the last change stream document
      and passed to the ``Watch()`` method as an option. Passing the ``resumeToken``
      to the ``Watch()`` method directs
      the change stream to attempt to resume notifications starting after the
      operation specified in the resume token.



      .. literalinclude:: /driver-examples/ChangeStreamExamples.cs
         :language: csharp
         :dedent: 14
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: c

      In the example below, the ``resumeAfter`` option is appended to the stream options
      to recreate the stream after it has been destroyed. Passing the ``_id`` to
      the change stream attempts to resume notifications starting after the
      operation specified.
   

      .. literalinclude:: /driver-examples/test-mongoc-sample-commands.c
         :language: C
         :dedent: 3
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3


   .. tab::
      :tabid: nodejs

      You can use the ``resumeAfter`` option to resume
      notifications after the operation specified in the resume
      token. The ``resumeAfter`` option takes a value that must
      resolve to a resume token, e.g. ``resumeToken`` in the
      example below.

      .. literalinclude:: /driver-examples/node_changestreams.js
         :language: javascript
         :dedent: 6
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: php

      You can use the ``resumeAfter`` option to resume
      notifications after the operation specified in the resume
      token. The ``resumeAfter`` option takes a value that must
      resolve to a resume token, e.g. ``$resumeToken`` in the
      example below.

      .. literalinclude:: /driver-examples/DocumentationExamplesTest.php
         :language: php
         :dedent: 8
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: ruby

      You can use the ``resume_after`` modifier to resume
      notifications after the operation specified in the resume
      token. The ``resume_after`` modifier takes a value that must
      resolve to a resume token, e.g. ``resume_token`` in the
      example below.

      .. literalinclude:: /driver-examples/change_stream_examples_spec.rb
         :language: ruby
         :dedent: 6
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: go

      You can use `ChangeStreamOptions.SetResumeAfter
      <https://godoc.org/go.mongodb.org/mongo-driver/mongo/options#ChangeStreamOptions.SetResumeAfter>`_
      to specify the resume
      token for the change stream. If the resumeAfter option is set,
      the change stream resumes notifications after the operation
      specified in the resume token. The ``SetResumeAfter`` takes a
      value that must resolve to a resume token, e.g.
      ``resumeToken`` in the example below.

      .. literalinclude:: /driver-examples/go_examples.go
         :language: go
         :dedent: 2
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: swift-sync

      You can use the ``resumeAfter`` option to resume
      notifications after the operation specified in the resume
      token. The ``resumeAfter`` option takes a value that must
      resolve to a resume token, e.g. ``resumeToken`` in the
      example below.

      .. literalinclude:: /driver-examples/swiftSync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

   .. tab::
      :tabid: swift-async

      You can use the ``resumeAfter`` option to resume
      notifications after the operation specified in the resume
      token. The ``resumeAfter`` option takes a value that must
      resolve to a resume token, e.g. ``resumeToken`` in the
      example below.

      .. literalinclude:: /driver-examples/swiftAsync.swift
         :language: swift
         :dedent: 8
         :start-after: Start Changestream Example 3
         :end-before: End Changestream Example 3

.. _change-stream-start-after:

``startAfter`` for Change Streams
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

You can start a new change stream after a specific event by passing a resume
token to ``startAfter`` when opening the cursor. Unlike
:ref:`resumeAfter <change-stream-resume-after>`, ``startAfter`` can
resume notifications after an :ref:`invalidate event <change-event-invalidate>`
by creating a new change stream. 

See :ref:`change-stream-resume-token` for more information on the resume token.

.. important::

   - The oplog must have enough history to locate the operation
     associated with the token or the timestamp, if the timestamp is in
     the past.

.. _change-stream-resume-token:

Resume Tokens
~~~~~~~~~~~~~

The resume token is available from multiple sources:

.. list-table::
   :header-rows: 1
   :stub-columns: 1
   :widths: 20 80

   * - Source
     - Description

   * - :ref:`Change Events <change-event-resume-token>`
     - Each change event notification includes a resume token
       on the ``_id`` field.

   * - :ref:`Aggregation <aggregate-resume-token>`
     - The :pipeline:`$changeStream` aggregation stage includes 
       a resume token on the ``cursor.postBatchResumeToken`` field.

       This field only appears when using the :dbcommand:`aggregate`
       command.

   * - :ref:`Get More <getMore-resume-token>`
     - The :dbcommand:`getMore` command includes a resume token on the
       ``cursor.postBatchResumeToken`` field.

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

.. tip::

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

.. _change-event-resume-token:

Resume Tokens from Change Events
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Change event notifications include a resume token on the ``_id`` field:

.. code-block:: json
   :copyable: false
   :emphasize-lines: 2-4


   {
      "_id": {
         "_data": "82635019A0000000012B042C0100296E5A1004AB1154ACACD849A48C61756D70D3B21F463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F69640064635019A078BE67426D7CF4D2000004"
       },
       "operationType": "insert",
       "clusterTime": Timestamp({ "t": 1666193824, "i": 1 }),
       "collectionUUID": new UUID("ab1154ac-acd8-49a4-8c61-756d70d3b21f"),
       "wallTime": ISODate("2022-10-19T15:37:04.604Z"),
       "fullDocument": {
          "_id": ObjectId("635019a078be67426d7cf4d2"'),
          "name": "Giovanni Verga"
       },
       "ns": { 
          "db": "test", 
          "coll": "names" 
       },
       "documentKey": { 
          "_id": ObjectId("635019a078be67426d7cf4d2") 
       }
   }


.. _aggregate-resume-token:

Resume Tokens from ``aggregate``
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

When using the :dbcommand:`aggregate` command, the :pipeline:`$changeStream`
aggregation stage includes a resume token on the  
``cursor.postBatchResumeToken`` field:

.. code-block:: json
   :copyable: false
   :emphasize-lines: 4-6

   {
      "cursor": {
         "firstBatch": [],
         "postBatchResumeToken": { 
            "_data": "8263515EAC000000022B0429296E1404" 
         },
         "id": Long("4309380460777152828"),
         "ns": "test.names"
      },
      "ok": 1,
      "$clusterTime": {
         "clusterTime": Timestamp({ "t": 1666277036, "i": 1 }),
         "signature": {
            "hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
            "keyId": Long("0")
         }
      },
      "operationTime": Timestamp({ "t": 1666277036, "i": 1 })
   }

.. _getMore-resume-token:

Resume Tokens from ``getMore``
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

The :dbcommand:`getMore` command also includes a resume token on the 
``cursor.postBatchResumeToken`` field:

.. code-block:: json
   :copyable: false
   :emphasize-lines: 4-6

   {
      "cursor": {
         "nextBatch": [],
         "postBatchResumeToken": { 
            "_data": "8263515979000000022B0429296E1404"
         },
         "id": Long("7049907285270685005"),
         "ns": "test.names"
      }, 
      "ok": 1,
      "$clusterTime": {
         "clusterTime": Timestamp( { "t": 1666275705, "i": 1 } ),
         "signature": {
            "hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
            "keyId": Long("0")
         }
      },
      "operationTime": Timestamp({ "t": 1666275705, "i": 1 })
   }


Use Cases
---------

Change streams can benefit architectures with reliant business systems,
informing downstream systems once data changes are durable. For example,
change streams can save time for developers when implementing Extract,
Transform, and Load (ETL) services, cross-platform synchronization,
collaboration functionality, and notification services.

.. _change-streams-access:

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

For deployments enforcing :ref:`authentication` and :ref:`authorization
<authorization>`:

- To open a change stream against specific collection, applications
  must have privileges that grant :authaction:`changeStream` and
  :authaction:`find` actions on the corresponding collection.

  .. code-block:: javascript

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

- To open a change stream on a single database, applications must have
  privileges that grant :authaction:`changeStream` and
  :authaction:`find` actions on all non-``system`` collections in the
  database.

  .. code-block:: javascript

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

- To open a change stream on an entire deployment, applications must
  have privileges that grant :authaction:`changeStream` and
  :authaction:`find` actions on all non-``system`` collections for all
  databases in the deployment.

  .. code-block:: javascript

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

.. _change-streams-event-notification:

Event Notification
------------------

Change streams only notify on data changes that have persisted to a majority
of data-bearing members in the replica set. This ensures that notifications are
triggered only by majority-committed changes that are durable in failure
scenarios.

For example, consider a 3-member :term:`replica set` with a change stream
cursor opened against the :term:`primary`. If a client issues an insert
operation, the change stream only notifies the application of the data change
once that insert has persisted to a majority of data-bearing members.

If an operation is associated with a :doc:`transaction
</core/transactions>`, the change event document includes the
``txnNumber`` and the ``lsid``.

Collation
---------

Change streams use ``simple`` binary comparisons unless an explicit collation is 
provided. 

Change Streams and Orphan Documents
-----------------------------------

.. include:: /includes/change-streams-and-orphans.rst

.. _change-stream-pre-post-images:

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

For complete examples with the change stream output, see
:ref:`db.collection.watch-change-streams-pre-and-post-images-example`.


.. toctree::
   :titlesonly:
   :hidden:

   Production Recommendations </administration/change-streams-production-recommendations>
   Change Events </reference/change-events>