CDRIVER-4362 Add showExpandedEvents option for C2C replication (#1133)

CDRIVER-4362 Add support for dropTarget, showExpandedEvents and dropIndex

Author: galon1

build/generate-opts.py

@@ -309,6 +309,7 @@ def __init__(self, items, **defaults):
                     'returns the value of the document before the associated '
                     'change.',
         }),
+        ('showExpandedEvents', { 'type': 'bool', 'help': 'Set to ``true`` to return an expanded list of change stream events. Avaiable only on MongoDB versions >=6.0'}),
         comment_option_string_pre_4_4,
     ], fullDocument=None, fullDocumentBeforeChange=None)),
 

src/libmongoc/doc/includes/change-stream-opts.txt

@@ -12,4 +12,5 @@
 * ``maxAwaitTimeMS``: An ``int64`` representing the maximum amount of time a call to :symbol:`mongoc_change_stream_next` will block waiting for data
 * ``fullDocument``: An optional UTF-8 string. Set this option to "default", "updateLookup", "whenAvailable", or "required", If unset, The string "default" is assumed. Set this 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.
 * ``fullDocumentBeforeChange``: An optional UTF-8 string. Set this option to "whenAvailable", "required", or "off". When unset, the default value is "off". Similar to "fullDocument", but returns the value of the document before the associated change.
+* ``showExpandedEvents``: Set to ``true`` to return an expanded list of change stream events. Avaiable only on MongoDB versions >=6.0
 * ``comment``: A :symbol:`bson_value_t` specifying the comment to attach to this command. The comment will appear in log messages, profiler output, and currentOp output. Only string values are supported prior to MongoDB 4.4.

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

@@ -39,6 +39,7 @@ struct _mongoc_change_stream_t {
    bson_t resume_token;
    bson_t *full_document;
    bson_t *full_document_before_change;
+   bool show_expanded_events;
 
    bson_error_t err;
    bson_t err_doc;

src/libmongoc/src/mongoc/mongoc-change-stream.c

@@ -129,6 +129,11 @@ _make_command (mongoc_change_stream_t *stream, bson_t *command)
    if (stream->full_document_before_change) {
       bson_concat (&change_stream_doc, stream->full_document_before_change);
    }
+   if (stream->show_expanded_events) {
+      BSON_APPEND_BOOL (&change_stream_doc,
+                        "showExpandedEvents",
+                        stream->show_expanded_events);
+   }
 
    if (stream->resumed) {
       /* Change stream spec: Resume Process */
@@ -433,6 +438,7 @@ _change_stream_init (mongoc_change_stream_t *stream,
 
    stream->batch_size = stream->opts.batchSize;
    stream->max_await_time_ms = stream->opts.maxAwaitTimeMS;
+   stream->show_expanded_events = stream->opts.showExpandedEvents;
 
    /* Accept two forms of user pipeline:
     * 1. A document like: { "pipeline": [...] }

src/libmongoc/src/mongoc/mongoc-opts-private.h

@@ -146,6 +146,7 @@ typedef struct _mongoc_change_stream_opts_t {
    int64_t maxAwaitTimeMS;
    const char *fullDocument;
    const char *fullDocumentBeforeChange;
+   bool showExpandedEvents;
    bson_value_t comment;
    bson_t extra;
 } mongoc_change_stream_opts_t;

src/libmongoc/src/mongoc/mongoc-opts.c

@@ -1583,6 +1583,7 @@ _mongoc_change_stream_opts_parse (
    mongoc_change_stream_opts->maxAwaitTimeMS = 0;
    mongoc_change_stream_opts->fullDocument = NULL;
    mongoc_change_stream_opts->fullDocumentBeforeChange = NULL;
+   mongoc_change_stream_opts->showExpandedEvents = false;
    memset (&mongoc_change_stream_opts->comment, 0, sizeof (bson_value_t));
    bson_init (&mongoc_change_stream_opts->extra);
 
@@ -1662,6 +1663,15 @@ _mongoc_change_stream_opts_parse (
             return false;
          }
       }
+      else if (!strcmp (bson_iter_key (&iter), "showExpandedEvents")) {
+         if (!_mongoc_convert_bool (
+               client,
+               &iter,
+               &mongoc_change_stream_opts->showExpandedEvents,
+               error)) {
+            return false;
+         }
+      }
       else if (!strcmp (bson_iter_key (&iter), "comment")) {
          if (!_mongoc_convert_bson_value_t (
                client,