CDRIVER-2670 add db and client watch

Author: kevinAlbs

src/libmongoc/doc/mongoc_change_stream_t.rst

@@ -32,6 +32,8 @@ Example
     :titlesonly:
     :maxdepth: 1
 
+    mongoc_client_watch
+    mongoc_database_watch
     mongoc_collection_watch
     mongoc_change_stream_next
     mongoc_change_stream_error_document

src/libmongoc/doc/mongoc_client_t.rst

@@ -83,5 +83,6 @@ Example
     mongoc_client_set_stream_initiator
     mongoc_client_set_write_concern
     mongoc_client_start_session
+    mongoc_client_watch
     mongoc_client_write_command_with_opts
 

src/libmongoc/doc/mongoc_client_watch.rst

@@ -0,0 +1,55 @@
+:man_page: mongoc_client_watch
+
+mongoc_client_watch()
+=====================
+
+Synopsis
+--------
+
+.. code-block:: c
+
+  mongoc_change_stream_t*
+  mongoc_client_watch (mongoc_client_t *client,
+                       const bson_t *pipeline,
+                       const bson_t *opts);
+
+A helper function to create a change stream. It is preferred to call this
+function over using a raw aggregation to create a change stream.
+
+This function uses the read preference and read concern of the client. If
+the change stream needs to re-establish connection, the same read preference
+will be used. This may happen if the change stream encounters a resumable error.
+
+.. warning::
+
+   A change stream is only supported with majority read concern.
+
+Parameters
+----------
+
+* ``db``: A :symbol:`mongoc_client_t` specifying the client which the change stream listens to.
+* ``pipeline``: A :symbol:`bson:bson_t` representing an aggregation pipeline appended to the change stream. This may be an empty document.
+* ``opts``: A :symbol:`bson:bson_t` containing change stream options or ``NULL``.
+
+``opts`` may be ``NULL`` or a document consisting of any subset of the following
+parameters:
+
+* ``batchSize`` An ``int32`` representing number of documents requested to be returned on each call to :symbol:`mongoc_change_stream_next`
+* ``resumeAfter`` A ``Document`` representing the logical starting point of the change stream. The ``_id`` field  of any change received from a change stream can be used here.
+* ``startAtOperationTime`` A ``Timestamp``. The change stream only provides changes that occurred at or after the specified timestamp. Any command run against the server will return an operation time that can be used here.
+* ``maxAwaitTimeMS`` An ``int64`` representing the maximum amount of time a call to :symbol:`mongoc_change_stream_next` will block waiting for data
+* ``collation`` A `Collation Document <https://docs.mongodb.com/manual/reference/collation/>`_
+
+Returns
+-------
+A newly allocated :symbol:`mongoc_change_stream_t` which must be freed with
+:symbol:`mongoc_change_stream_destroy` when no longer in use. The returned
+:symbol:`mongoc_change_stream_t` is never ``NULL``. If there is an error, it can
+be retrieved with :symbol:`mongoc_change_stream_error_document`, and subsequent
+calls to :symbol:`mongoc_change_stream_next` will return ``false``.
+
+See Also
+--------
+:doc:`mongoc_database_watch`
+
+:doc:`mongoc_collection_watch`

src/libmongoc/doc/mongoc_collection_watch.rst

@@ -35,7 +35,8 @@ Parameters
 parameters:
 
 * ``batchSize`` An ``int32`` representing number of documents requested to be returned on each call to :symbol:`mongoc_change_stream_next`
-* ``resumeAfter`` A ``Document`` representing the starting point of the change stream
+* ``resumeAfter`` A ``Document`` representing the logical starting point of the change stream. The ``_id`` field  of any change received from a change stream can be used here.
+* ``startAtOperationTime`` A ``Timestamp``. The change stream only provides changes that occurred at or after the specified timestamp. Any command run against the server will return an operation time that can be used here.
 * ``maxAwaitTimeMS`` An ``int64`` representing the maximum amount of time a call to :symbol:`mongoc_change_stream_next` will block waiting for data
 * ``collation`` A `Collation Document <https://docs.mongodb.com/manual/reference/collation/>`_
 
@@ -45,4 +46,10 @@ A newly allocated :symbol:`mongoc_change_stream_t` which must be freed with
 :symbol:`mongoc_change_stream_destroy` when no longer in use. The returned
 :symbol:`mongoc_change_stream_t` is never ``NULL``. If there is an error, it can
 be retrieved with :symbol:`mongoc_change_stream_error_document`, and subsequent
-calls to :symbol:`mongoc_change_stream_next` will return ``false``.
+calls to :symbol:`mongoc_change_stream_next` will return ``false``.
+
+See Also
+--------
+:doc:`mongoc_client_watch`
+
+:doc:`mongoc_database_watch`

src/libmongoc/doc/mongoc_database_t.rst

@@ -51,6 +51,7 @@ Read preferences and write concerns are inherited from the parent client. They c
     mongoc_database_set_read_concern
     mongoc_database_set_read_prefs
     mongoc_database_set_write_concern
+    mongoc_database_watch
     mongoc_database_write_command_with_opts
 
 Examples

src/libmongoc/doc/mongoc_database_watch.rst

@@ -0,0 +1,55 @@
+:man_page: mongoc_database_watch
+
+mongoc_database_watch()
+=======================
+
+Synopsis
+--------
+
+.. code-block:: c
+
+  mongoc_change_stream_t*
+  mongoc_database_watch (const mongoc_database_t *db,
+                         const bson_t *pipeline,
+                         const bson_t *opts);
+
+A helper function to create a change stream. It is preferred to call this
+function over using a raw aggregation to create a change stream.
+
+This function uses the read preference and read concern of the database. If
+the change stream needs to re-establish connection, the same read preference
+will be used. This may happen if the change stream encounters a resumable error.
+
+.. warning::
+
+   A change stream is only supported with majority read concern.
+
+Parameters
+----------
+
+* ``db``: A :symbol:`mongoc_database_t` specifying the database which the change stream listens to.
+* ``pipeline``: A :symbol:`bson:bson_t` representing an aggregation pipeline appended to the change stream. This may be an empty document.
+* ``opts``: A :symbol:`bson:bson_t` containing change stream options or ``NULL``.
+
+``opts`` may be ``NULL`` or a document consisting of any subset of the following
+parameters:
+
+* ``batchSize`` An ``int32`` representing number of documents requested to be returned on each call to :symbol:`mongoc_change_stream_next`
+* ``resumeAfter`` A ``Document`` representing the logical starting point of the change stream. The ``_id`` field  of any change received from a change stream can be used here.
+* ``startAtOperationTime`` A ``Timestamp``. The change stream only provides changes that occurred at or after the specified timestamp. Any command run against the server will return an operation time that can be used here.
+* ``maxAwaitTimeMS`` An ``int64`` representing the maximum amount of time a call to :symbol:`mongoc_change_stream_next` will block waiting for data
+* ``collation`` A `Collation Document <https://docs.mongodb.com/manual/reference/collation/>`_
+
+Returns
+-------
+A newly allocated :symbol:`mongoc_change_stream_t` which must be freed with
+:symbol:`mongoc_change_stream_destroy` when no longer in use. The returned
+:symbol:`mongoc_change_stream_t` is never ``NULL``. If there is an error, it can
+be retrieved with :symbol:`mongoc_change_stream_error_document`, and subsequent
+calls to :symbol:`mongoc_change_stream_next` will return ``false``.
+
+See Also
+--------
+:doc:`mongoc_client_watch`
+
+:doc:`mongoc_collection_watch`

src/libmongoc/src/mongoc/mongoc-change-stream-private.h

@@ -22,26 +22,51 @@
 #include "mongoc-collection.h"
 #include "mongoc-cursor.h"
 
+typedef enum {
+   MONGOC_CHANGE_STREAM_COLLECTION,
+   MONGOC_CHANGE_STREAM_DATABASE,
+   MONGOC_CHANGE_STREAM_CLIENT
+} mongoc_change_stream_type_t;
+
 struct _mongoc_change_stream_t {
    bson_t pipeline_to_append;
    bson_t full_document;
    bson_t opts;
-   bson_t resume_token; /* empty, or has resumeAfter: doc */
+   bson_t resume_token;   /* empty, or has resumeAfter: doc */
+   bson_t operation_time; /* empty, or has startAtOperationTime: ts */
 
    bson_error_t err;
    bson_t err_doc;
 
    mongoc_cursor_t *cursor;
-   mongoc_collection_t *coll;
+
+   mongoc_client_t *client;
+   mongoc_read_prefs_t *read_prefs;
+   mongoc_read_concern_t *read_concern;
+
+   mongoc_change_stream_type_t change_stream_type;
+   char db[140];
+   char coll[140];
+
    int64_t max_await_time_ms;
    int32_t batch_size;
 
    mongoc_client_session_t *implicit_session;
 };
 
 mongoc_change_stream_t *
-_mongoc_change_stream_new (const mongoc_collection_t *coll,
-                           const bson_t *pipeline,
-                           const bson_t *opts);
+_mongoc_change_stream_new_from_collection (const mongoc_collection_t *coll,
+                                           const bson_t *pipeline,
+                                           const bson_t *opts);
+
+mongoc_change_stream_t *
+_mongoc_change_stream_new_from_database (const mongoc_database_t *db,
+                                         const bson_t *pipeline,
+                                         const bson_t *opts);
+
+mongoc_change_stream_t *
+_mongoc_change_stream_new_from_client (mongoc_client_t *client,
+                                       const bson_t *pipeline,
+                                       const bson_t *opts);
 
 #endif /* MONGOC_CHANGE_STREAM_PRIVATE_H */