CDRIVER-2606 array arg on collection_watch

Author: kevinAlbs

src/mongoc/mongoc-change-stream.c

@@ -271,10 +271,22 @@ _mongoc_change_stream_new (const mongoc_collection_t *coll,
       }
    }
 
+   /* Accept two forms of user pipeline:
+    * 1. A document like: { "pipeline": [...] }
+    * 2. An array-like document: { "0": {}, "1": {}, ... }
+    * If the passed pipeline is invalid, we pass it along and let the server
+    * error instead.
+    */
    if (!bson_empty (pipeline)) {
       bson_iter_t iter;
-      if (bson_iter_init_find (&iter, pipeline, "pipeline")) {
+      if (bson_iter_init_find (&iter, pipeline, "pipeline") &&
+          BSON_ITER_HOLDS_ARRAY (&iter)) {
          SET_BSON_OR_ERR (&stream->pipeline_to_append, "pipeline");
+      } else {
+         if (!BSON_APPEND_ARRAY (
+                &stream->pipeline_to_append, "pipeline", pipeline)) {
+            CHANGE_STREAM_ERR ("pipeline");
+         }
       }
    }
 

tests/test-mongoc-change-stream.c

@@ -1095,6 +1095,97 @@ typedef struct {
    int agg_count;
 } array_started_ctx_t;
 
+static void
+_accepts_array_started (const mongoc_apm_command_started_t *event)
+{
+   const bson_t *cmd = mongoc_apm_command_started_get_command (event);
+   const char *cmd_name = mongoc_apm_command_started_get_command_name (event);
+   array_started_ctx_t *ctx =
+      (array_started_ctx_t *) mongoc_apm_command_started_get_context (event);
+   if (strcmp (cmd_name, "aggregate") != 0) {
+      return;
+   }
+   ctx->agg_count++;
+   ASSERT_MATCH (cmd, ctx->pattern);
+}
+
+/* Test that watch accepts an array document {0: {}, 1: {}} as the pipeline,
+ * similar to mongoc_collection_aggregate */
+static void
+test_change_stream_accepts_array (void *test_ctx)
+{
+   mongoc_client_t *client = test_framework_client_new ();
+   mongoc_apm_callbacks_t *callbacks = mongoc_apm_callbacks_new ();
+   array_started_ctx_t ctx = {0};
+   mongoc_collection_t *coll;
+   mongoc_change_stream_t *stream;
+   const bson_t *bson;
+   bson_error_t err;
+   bson_t *opts =
+      tmp_bson ("{'maxAwaitTimeMS': 1}"); /* to speed up the test. */
+
+   mongoc_client_set_error_api (client, MONGOC_ERROR_API_VERSION_2);
+   /* set up apm callbacks to listen for the agg commands. */
+   ctx.pattern =
+      bson_strdup ("{'aggregate': 'coll', 'pipeline': [ {'$changeStream': {}}, "
+                   "{'$match': {'x': 1}}, {'$project': {'x': 1}}]}");
+   mongoc_apm_set_command_started_cb (callbacks, _accepts_array_started);
+   mongoc_client_set_apm_callbacks (client, callbacks, &ctx);
+   coll = mongoc_client_get_collection (client, "db", "coll");
+   ASSERT_OR_PRINT (
+      mongoc_collection_insert_one (coll, tmp_bson (NULL), NULL, NULL, &err),
+      err);
+   /* try starting a change stream with a { "pipeline": [...] } argument */
+   stream = mongoc_collection_watch (
+      coll,
+      tmp_bson ("{'pipeline': [{'$match': {'x': 1}}, {'$project': {'x': 1}}]}"),
+      opts);
+   (void) mongoc_change_stream_next (stream, &bson);
+   ASSERT_OR_PRINT (!mongoc_change_stream_error_document (stream, &err, &bson),
+                    err);
+   ASSERT_CMPINT32 (ctx.agg_count, ==, 1);
+   mongoc_change_stream_destroy (stream);
+   /* try with an array like document. */
+   stream = mongoc_collection_watch (
+      coll,
+      tmp_bson ("{'0': {'$match': {'x': 1}}, '1': {'$project': {'x': 1}}}"),
+      opts);
+   (void) mongoc_change_stream_next (stream, &bson);
+   ASSERT_OR_PRINT (!mongoc_change_stream_error_document (stream, &err, &bson),
+                    err);
+   ASSERT_CMPINT32 (ctx.agg_count, ==, 2);
+   mongoc_change_stream_destroy (stream);
+   /* try with malformed { "pipeline": [...] } argument. */
+   bson_free (ctx.pattern);
+   ctx.pattern = bson_strdup (
+      "{'aggregate': 'coll', 'pipeline': [ {'$changeStream': {}}, 42 ]}");
+   stream =
+      mongoc_collection_watch (coll, tmp_bson ("{'pipeline': [42] }"), NULL);
+   (void) mongoc_change_stream_next (stream, &bson);
+   BSON_ASSERT (mongoc_change_stream_error_document (stream, &err, &bson));
+   ASSERT_ERROR_CONTAINS (
+      err,
+      MONGOC_ERROR_SERVER,
+      14,
+      "Each element of the 'pipeline' array must be an object");
+   ASSERT_CMPINT32 (ctx.agg_count, ==, 3);
+   mongoc_change_stream_destroy (stream);
+   /* try with malformed array doc argument. */
+   stream = mongoc_collection_watch (coll, tmp_bson ("{'0': 42 }"), NULL);
+   (void) mongoc_change_stream_next (stream, &bson);
+   BSON_ASSERT (mongoc_change_stream_error_document (stream, &err, &bson));
+   ASSERT_ERROR_CONTAINS (
+      err,
+      MONGOC_ERROR_SERVER,
+      14,
+      "Each element of the 'pipeline' array must be an object");
+   ASSERT_CMPINT32 (ctx.agg_count, ==, 4);
+   mongoc_change_stream_destroy (stream);
+   bson_free (ctx.pattern);
+   mongoc_apm_callbacks_destroy (callbacks);
+   mongoc_collection_destroy (coll);
+   mongoc_client_destroy (client);
+}
 
 void
 test_change_stream_install (TestSuite *suite)
@@ -1165,4 +1256,11 @@ test_change_stream_install (TestSuite *suite)
                       NULL,
                       NULL,
                       test_framework_skip_if_not_rs_version_6);
+
+   TestSuite_AddFull (suite,
+                      "/change_stream/accepts_array",
+                      test_change_stream_accepts_array,
+                      NULL,
+                      NULL,
+                      test_framework_skip_if_not_rs_version_6);
 }

tests/test-mongoc-sample-commands.c

@@ -2598,7 +2598,6 @@ test_example_change_stream (mongoc_database_t *db)
    /* End Changestream Example 3 */
    bson_destroy (&opts);
    bson_destroy (&pipeline);
-   bson_destroy (&opts);
    mongoc_collection_destroy (collection);
 }