Merge pull request #14348 from Automattic/vkarpov15/gh-14331

fix(cursor): make aggregation cursor support `transform` option to match query cursor

Author: vkarpov15

lib/cursor/aggregationCursor.js

@@ -57,12 +57,20 @@ util.inherits(AggregationCursor, Readable);
 function _init(model, c, agg) {
   if (!model.collection.buffer) {
     model.hooks.execPre('aggregate', agg, function() {
+      if (typeof agg.options?.cursor?.transform === 'function') {
+        c._transforms.push(agg.options.cursor.transform);
+      }
+
       c.cursor = model.collection.aggregate(agg._pipeline, agg.options || {});
       c.emit('cursor', c.cursor);
     });
   } else {
     model.collection.emitter.once('queue', function() {
       model.hooks.execPre('aggregate', agg, function() {
+        if (typeof agg.options?.cursor?.transform === 'function') {
+          c._transforms.push(agg.options.cursor.transform);
+        }
+
         c.cursor = model.collection.aggregate(agg._pipeline, agg.options || {});
         c.emit('cursor', c.cursor);
       });

test/aggregate.test.js

@@ -7,6 +7,7 @@
 const start = require('./common');
 
 const assert = require('assert');
+const stream = require('stream');
 
 const Aggregate = require('../lib/aggregate');
 
@@ -1215,6 +1216,32 @@ describe('aggregate: ', function() {
     assert.equal(res[1].test, 'a test');
   });
 
+  it('cursor supports transform option (gh-14331)', async function() {
+    const mySchema = new Schema({ name: String });
+    const Test = db.model('Test', mySchema);
+
+    await Test.deleteMany({});
+    await Test.create([{ name: 'Apple' }, { name: 'Apple' }]);
+
+    let resolve;
+    const waitForStream = new Promise(innerResolve => {
+      resolve = innerResolve;
+    });
+    const otherStream = new stream.Writable({
+      write(chunk, encoding, callback) {
+        resolve(chunk.toString());
+        callback();
+      }
+    });
+
+    await Test.
+      aggregate([{ $match: { name: 'Apple' } }]).
+      cursor({ transform: JSON.stringify }).
+      pipe(otherStream);
+    const streamValue = await waitForStream;
+    assert.ok(streamValue.includes('"name":"Apple"'), streamValue);
+  });
+
   describe('Mongo 3.6 options', function() {
     before(async function() {
       await onlyTestAtOrAbove('3.6', this);